From b570f29a21939b4eab92d8635baea00bb73175d9 Mon Sep 17 00:00:00 2001 From: tchivs Date: Mon, 7 Sep 2026 09:13:57 +0800 Subject: [PATCH 1/3] [FLINK-40594][cdc-common] Preserve TIME precision in the pipeline runtime TimeType accepts a precision of up to 9 and the Debezium deserializer already produces microsecond and nanosecond values, but the runtime stored TIME as millisecond-of-day. Every TIME(p) with p > 3 was silently truncated, and two values differing only below the millisecond compared equal, collapsing them during deduplication, ordering and key comparison. TimeData now keeps nanosecond-of-day and exposes toMicroOfDay/toNanoOfDay; toMillisOfDay keeps its signature and truncating behaviour. TimeDataSerializer becomes precision-aware like TimestampDataSerializer: precision <= 3 keeps the historical four-byte millisecond encoding and stays compatible as is, while precision > 3 uses an eight-byte nanosecond encoding reached through a versioned snapshot that reports compatible-after-migration and still reads the legacy snapshot envelope and four-byte payload. The binary writers/readers and InternalSerializers thread the declared precision through. Two existing expectations pinned the truncation and are corrected rather than relaxed: the TIME(6)/TIME(9) transform expectation now asserts the retained fraction, and the temporal-function check compares LOCALTIME at whole-second granularity because LOCALTIME and CURRENT_TIME are TIME(0). --- .../cdc/common/converter/CommonConverter.java | 29 ++- .../converter/InternalObjectConverter.java | 2 +- .../common/converter/JavaObjectConverter.java | 10 +- .../flink/cdc/common/data/ArrayData.java | 15 +- .../cdc/common/data/GenericArrayData.java | 10 + .../flink/cdc/common/data/RecordData.java | 14 +- .../flink/cdc/common/data/TimeData.java | 39 ++-- .../common/data/binary/BinaryArrayData.java | 20 +- .../common/data/binary/BinaryRecordData.java | 15 ++ .../InternalObjectConverterTest.java | 6 + .../converter/JavaObjectConverterTest.java | 6 + .../flink/cdc/common/data/TimeDataTest.java | 53 +++++ .../flink/FlinkPipelineTransformITCase.java | 9 +- .../source/PostgresFullTypesITCase.java | 5 + .../serializer/InternalSerializers.java | 2 +- .../serializer/data/TimeDataSerializer.java | 160 +++++++++++++-- .../data/writer/AbstractBinaryWriter.java | 6 +- .../data/writer/BinaryArrayWriter.java | 14 +- .../transform/PostTransformOperatorTest.java | 2 +- .../data/ArrayDataSerializerTest.java | 22 ++ .../data/TimeDataSerializerTest.java | 191 ++++++++++++++++-- .../BinaryRecordDataExtractorTest.java | 8 +- .../BinaryRecordDataGeneratorTest.java | 26 +++ 23 files changed, 588 insertions(+), 76 deletions(-) create mode 100644 flink-cdc-common/src/test/java/org/apache/flink/cdc/common/data/TimeDataTest.java diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/converter/CommonConverter.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/converter/CommonConverter.java index 8260c305d00..d37a152ae98 100644 --- a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/converter/CommonConverter.java +++ b/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; diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/converter/InternalObjectConverter.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/converter/InternalObjectConverter.java index 0dc664e54ba..dd1ca1a0d5b 100644 --- a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/converter/InternalObjectConverter.java +++ b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/converter/InternalObjectConverter.java @@ -127,7 +127,7 @@ public Function visit(DateType dateType) { @Override public Function visit(TimeType timeType) { - return CommonConverter::convertToTimeData; + return value -> CommonConverter.convertToTimeData(value, timeType.getPrecision()); } @Override diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/converter/JavaObjectConverter.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/converter/JavaObjectConverter.java index ddb848a1669..be05fa37efc 100644 --- a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/converter/JavaObjectConverter.java +++ b/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 visit(DateType dateType) { @Override public Function 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 diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/ArrayData.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/ArrayData.java index 8b25b4586bf..721e2604cc9 100644 --- a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/ArrayData.java +++ b/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. + * + *

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; diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/GenericArrayData.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/GenericArrayData.java index 22a2b9336a9..7f8badb13ac 100644 --- a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/GenericArrayData.java +++ b/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); diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/RecordData.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/RecordData.java index c0289ab6c09..9c0d63e36a8 100644 --- a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/RecordData.java +++ b/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. + * + *

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); diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/TimeData.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/TimeData.java index 54de38ee4af..083fcda9e68 100644 --- a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/TimeData.java +++ b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/TimeData.java @@ -26,31 +26,30 @@ */ public class TimeData implements Comparable { - 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); } } diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/binary/BinaryArrayData.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/binary/BinaryArrayData.java index 991085ae226..89f1d342478 100644 --- a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/binary/BinaryArrayData.java +++ b/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); diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/binary/BinaryRecordData.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/binary/BinaryRecordData.java index bc645eb7db4..2b9dc30f04e 100644 --- a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/binary/BinaryRecordData.java +++ b/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)); } diff --git a/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/converter/InternalObjectConverterTest.java b/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/converter/InternalObjectConverterTest.java index ee4ff5859d5..b8f8df99842 100644 --- a/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/converter/InternalObjectConverterTest.java +++ b/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(); } diff --git a/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/converter/JavaObjectConverterTest.java b/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/converter/JavaObjectConverterTest.java index 786ef63e2fc..5127914e025 100644 --- a/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/converter/JavaObjectConverterTest.java +++ b/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(); } diff --git a/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/data/TimeDataTest.java b/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/data/TimeDataTest.java new file mode 100644 index 00000000000..0e31ba4102c --- /dev/null +++ b/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()); + } +} diff --git a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineTransformITCase.java b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineTransformITCase.java index f1b89a94b21..cb5f07f15ca 100644 --- a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineTransformITCase.java +++ b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineTransformITCase.java @@ -3072,7 +3072,7 @@ void testDateAndTimeCastingFunctions() throws Exception { assertThat(outputEvents) .containsExactlyInAnyOrder( "CreateTableEvent{tableId=default_namespace.default_schema.my_table, schema=columns={`id` INT NOT NULL,`date_0` DATE,`time_0` TIME(0),`time_3` TIME(3),`time_6` TIME(6),`time_9` TIME(9),`date_0_str` STRING,`time_0_str` STRING,`time_3_str` STRING,`time_6_str` STRING,`time_9_str` STRING}, primaryKeys=id, options=()}", - "DataChangeEvent{tableId=default_namespace.default_schema.my_table, before=[], after=[1, 1999-12-31, 21:48:25, 21:48:25.123, 21:48:25.123, 21:48:25.123, 1999-12-31, 21:48:25, 21:48:25.123, 21:48:25.123, 21:48:25.123], op=INSERT, meta=()}", + "DataChangeEvent{tableId=default_namespace.default_schema.my_table, before=[], after=[1, 1999-12-31, 21:48:25, 21:48:25.123, 21:48:25.123456, 21:48:25.123456789, 1999-12-31, 21:48:25, 21:48:25.123, 21:48:25.123456, 21:48:25.123456789], op=INSERT, meta=()}", "DataChangeEvent{tableId=default_namespace.default_schema.my_table, before=[], after=[2, null, null, null, null, null, null, null, null, null, null], op=INSERT, meta=()}"); } @@ -3508,10 +3508,13 @@ void verifyDataRecord(String recordLine) { .toInstant(ZoneOffset.UTC); long milliSecondsInOneDay = 24 * 60 * 60 * 1000; + // LOCALTIME and CURRENT_TIME are TIME(0), so only whole seconds are part of their declared + // type. Comparing against the millisecond-of-day of CURRENT_TIMESTAMP would require the + // runtime to carry a fraction the column type does not have. assertThat(TimeData.fromIsoLocalTimeString(localTime)) .isEqualTo( - TimeData.fromMillisOfDay( - (int) (instant.toEpochMilli() % milliSecondsInOneDay))); + TimeData.fromSecondOfDay( + (int) (instant.toEpochMilli() % milliSecondsInOneDay / 1000))); String localDate = tokens.get(5); assertThat(DateData.fromIsoLocalDateString(localDate)) diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/test/java/org/apache/flink/cdc/connectors/postgres/source/PostgresFullTypesITCase.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/test/java/org/apache/flink/cdc/connectors/postgres/source/PostgresFullTypesITCase.java index e7561fdf1b6..beb5eb4be96 100644 --- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/test/java/org/apache/flink/cdc/connectors/postgres/source/PostgresFullTypesITCase.java +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/test/java/org/apache/flink/cdc/connectors/postgres/source/PostgresFullTypesITCase.java @@ -306,6 +306,11 @@ public void testTimeTypesWithTemporalModeAdaptive() throws Exception { RecordData snapshotRecord = ((DataChangeEvent) snapshotResults.get(0)).after(); Assertions.assertThat(recordFields(snapshotRecord, TIME_TYPES_WITH_ADAPTIVE)) .isEqualTo(expectedSnapshot); + // Without sub-millisecond retention the TIME(6) entry above and the value read back both + // truncate to 18:00:22.123, so the comparison passes while the microseconds are already + // gone. Assert the microsecond-of-day directly so the expectation cannot pass vacuously. + Assertions.assertThat(snapshotRecord.getTime(4, 6).toMicroOfDay()) + .isEqualTo(64_822_123_456L); } @Test diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/InternalSerializers.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/InternalSerializers.java index 8685f833314..a23ff99d1a1 100644 --- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/InternalSerializers.java +++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/InternalSerializers.java @@ -67,7 +67,7 @@ private static TypeSerializer createInternal(DataType type) { case DATE: return DateDataSerializer.INSTANCE; case TIME_WITHOUT_TIME_ZONE: - return TimeDataSerializer.INSTANCE; + return new TimeDataSerializer(getPrecision(type)); case BIGINT: return LongSerializer.INSTANCE; case FLOAT: diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/data/TimeDataSerializer.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/data/TimeDataSerializer.java index 414a7d964b7..64fa89e644b 100644 --- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/data/TimeDataSerializer.java +++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/data/TimeDataSerializer.java @@ -17,29 +17,55 @@ package org.apache.flink.cdc.runtime.serializer.data; -import org.apache.flink.api.common.typeutils.SimpleTypeSerializerSnapshot; +import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.api.common.typeutils.TypeSerializerSchemaCompatibility; import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot; +import org.apache.flink.api.common.typeutils.TypeSerializerSnapshotAdapter; import org.apache.flink.cdc.common.data.TimeData; -import org.apache.flink.cdc.runtime.serializer.TypeSerializerSingleton; import org.apache.flink.core.memory.DataInputView; import org.apache.flink.core.memory.DataOutputView; import java.io.IOException; +import java.io.ObjectInputStream; -/** Serializer for {@link TimeData}. */ -public final class TimeDataSerializer extends TypeSerializerSingleton { +/** + * Serializer for {@link TimeData}. + * + *

TIME values with precision up to 3 retain the historical four-byte millisecond encoding. + * Higher precisions use an eight-byte nanosecond-of-day encoding. + */ +public final class TimeDataSerializer extends TypeSerializer { private static final long serialVersionUID = 1L; - public static final TimeDataSerializer INSTANCE = new TimeDataSerializer(); + /** The historical singleton represents the millisecond TIME encoding. */ + public static final TimeDataSerializer INSTANCE = new TimeDataSerializer(3); + + private int precision; + private boolean legacyFormat; + + public TimeDataSerializer(int precision) { + this(precision, false); + } - private TimeDataSerializer() {} + private TimeDataSerializer(int precision, boolean legacyFormat) { + if (precision < 0 || precision > 9) { + throw new IllegalArgumentException("TIME precision must be between 0 and 9"); + } + this.precision = precision; + this.legacyFormat = legacyFormat; + } @Override public boolean isImmutableType() { return true; } + @Override + public TypeSerializer duplicate() { + return new TimeDataSerializer(precision, legacyFormat); + } + @Override public TimeData createInstance() { return TimeData.fromNanoOfDay(0); @@ -47,27 +73,33 @@ public TimeData createInstance() { @Override public TimeData copy(TimeData from) { - return TimeData.fromMillisOfDay(from.toMillisOfDay()); + return TimeData.fromNanoOfDay(from.toNanoOfDay()); } @Override public TimeData copy(TimeData from, TimeData reuse) { - return TimeData.fromMillisOfDay(from.toMillisOfDay()); + return copy(from); } @Override public int getLength() { - return 4; + return usesMillisEncoding() ? Integer.BYTES : Long.BYTES; } @Override public void serialize(TimeData record, DataOutputView target) throws IOException { - target.writeInt(record.toMillisOfDay()); + if (usesMillisEncoding()) { + target.writeInt(record.toMillisOfDay()); + } else { + target.writeLong(record.toNanoOfDay()); + } } @Override public TimeData deserialize(DataInputView source) throws IOException { - return TimeData.fromMillisOfDay(source.readInt()); + return usesMillisEncoding() + ? TimeData.fromMillisOfDay(source.readInt()) + : TimeData.fromNanoOfDay(source.readLong()); } @Override @@ -77,21 +109,117 @@ public TimeData deserialize(TimeData record, DataInputView source) throws IOExce @Override public void copy(DataInputView source, DataOutputView target) throws IOException { - target.writeInt(source.readInt()); + if (usesMillisEncoding()) { + target.writeInt(source.readInt()); + } else { + target.writeLong(source.readLong()); + } + } + + private boolean usesMillisEncoding() { + return legacyFormat || precision <= 3; + } + + @Override + public boolean equals(Object obj) { + if (this == obj) { + return true; + } + if (obj == null || getClass() != obj.getClass()) { + return false; + } + TimeDataSerializer that = (TimeDataSerializer) obj; + return precision == that.precision && legacyFormat == that.legacyFormat; + } + + @Override + public int hashCode() { + return 31 * precision + Boolean.hashCode(legacyFormat); } @Override public TypeSerializerSnapshot snapshotConfiguration() { - return new TimeDataSerializerSnapshot(); + return new TimeDataSerializerSnapshot(precision, legacyFormat); + } + + /** Reads Java-serialized serializer instances embedded in old array/map snapshots. */ + private void readObject(ObjectInputStream input) throws IOException, ClassNotFoundException { + ObjectInputStream.GetField fields = input.readFields(); + if (fields.defaulted("precision")) { + precision = 3; + legacyFormat = true; + } else { + precision = fields.get("precision", 3); + legacyFormat = fields.get("legacyFormat", false); + } } /** Serializer configuration snapshot for compatibility and format evolution. */ - @SuppressWarnings("WeakerAccess") public static final class TimeDataSerializerSnapshot - extends SimpleTypeSerializerSnapshot { + implements TypeSerializerSnapshotAdapter { + + // Versions 2 and 3 belonged to SimpleTypeSerializerSnapshot and contained no precision. + private static final int CURRENT_VERSION = 4; + + private int previousPrecision; + private boolean previousLegacyFormat; public TimeDataSerializerSnapshot() { - super(() -> INSTANCE); + // Used when restoring from a checkpoint/savepoint. + } + + private TimeDataSerializerSnapshot(int precision, boolean legacyFormat) { + this.previousPrecision = precision; + this.previousLegacyFormat = legacyFormat; + } + + @Override + public int getCurrentVersion() { + return CURRENT_VERSION; + } + + @Override + public void writeSnapshot(DataOutputView out) throws IOException { + out.writeInt(previousPrecision); + out.writeBoolean(previousLegacyFormat); + } + + @Override + public void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLoader) + throws IOException { + if (readVersion == 2) { + // SimpleTypeSerializerSnapshot v2 wrote its serializer class name. + in.readUTF(); + previousPrecision = 3; + previousLegacyFormat = true; + } else if (readVersion == 3) { + previousPrecision = 3; + previousLegacyFormat = true; + } else if (readVersion == CURRENT_VERSION) { + previousPrecision = in.readInt(); + previousLegacyFormat = in.readBoolean(); + } else { + throw new IOException( + "Unrecognized TimeDataSerializer snapshot version " + readVersion); + } + } + + @Override + public TypeSerializer restoreSerializer() { + return new TimeDataSerializer(previousPrecision, previousLegacyFormat); + } + + @Override + public TypeSerializerSchemaCompatibility resolveSchemaCompatibility( + TypeSerializer newSerializer) { + if (!(newSerializer instanceof TimeDataSerializer)) { + return TypeSerializerSchemaCompatibility.incompatible(); + } + TimeDataSerializer timeSerializer = (TimeDataSerializer) newSerializer; + boolean previousMillisEncoding = previousLegacyFormat || previousPrecision <= 3; + return previousMillisEncoding == timeSerializer.usesMillisEncoding() + ? TypeSerializerSchemaCompatibility.compatibleAsIs() + : TypeSerializerSchemaCompatibility.compatibleAfterMigration(); } } } diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/data/writer/AbstractBinaryWriter.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/data/writer/AbstractBinaryWriter.java index 5dc05bcddf4..ea07571da20 100644 --- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/data/writer/AbstractBinaryWriter.java +++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/data/writer/AbstractBinaryWriter.java @@ -236,7 +236,11 @@ public void writeDate(int pos, DateData value) { @Override public void writeTime(int pos, TimeData value, int precision) { - writeInt(pos, value.toMillisOfDay()); + if (precision <= 3) { + writeInt(pos, value.toMillisOfDay()); + } else { + writeLong(pos, Long.MIN_VALUE | value.toNanoOfDay()); + } } @Override diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/data/writer/BinaryArrayWriter.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/data/writer/BinaryArrayWriter.java index 1177bcdee44..92a8c57893b 100644 --- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/data/writer/BinaryArrayWriter.java +++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/data/writer/BinaryArrayWriter.java @@ -21,6 +21,7 @@ import org.apache.flink.cdc.common.data.binary.BinaryArrayData; import org.apache.flink.cdc.common.data.binary.BinarySegmentUtils; import org.apache.flink.cdc.common.types.DataType; +import org.apache.flink.cdc.common.types.DataTypeChecks; import org.apache.flink.core.memory.MemorySegmentFactory; import java.io.Serializable; @@ -126,9 +127,15 @@ public void setNullAt(int pos, DataType type) { break; case INTEGER: case DATE: - case TIME_WITHOUT_TIME_ZONE: setNullInt(pos); break; + case TIME_WITHOUT_TIME_ZONE: + if (DataTypeChecks.getPrecision(type) <= 3) { + setNullInt(pos); + } else { + setNullLong(pos); + } + break; case BIGINT: case TIMESTAMP_WITHOUT_TIME_ZONE: case TIMESTAMP_WITH_LOCAL_TIME_ZONE: @@ -248,8 +255,11 @@ public static BinaryArrayWriter.NullSetter createNullSetter(DataType elementType return BinaryArrayWriter::setNullShort; case INTEGER: case DATE: - case TIME_WITHOUT_TIME_ZONE: return BinaryArrayWriter::setNullInt; + case TIME_WITHOUT_TIME_ZONE: + return DataTypeChecks.getPrecision(elementType) <= 3 + ? BinaryArrayWriter::setNullInt + : BinaryArrayWriter::setNullLong; case FLOAT: return BinaryArrayWriter::setNullFloat; case DOUBLE: diff --git a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorTest.java b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorTest.java index 2e02b05a3b9..bf02feeba36 100644 --- a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorTest.java +++ b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorTest.java @@ -637,7 +637,7 @@ void testDataChangeEventTransformProjectionDataTypeConvert() throws Exception { 1, 1L, DateData.fromEpochDay(1704471599), - TimeData.fromMillisOfDay(1704471), + TimeData.fromSecondOfDay(1704), TimestampData.fromMillis(1704471599), 3.14f, 3.14d, diff --git a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/serializer/data/ArrayDataSerializerTest.java b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/serializer/data/ArrayDataSerializerTest.java index c03fe76f79c..6601b12bbf0 100644 --- a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/serializer/data/ArrayDataSerializerTest.java +++ b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/serializer/data/ArrayDataSerializerTest.java @@ -22,6 +22,7 @@ import org.apache.flink.cdc.common.data.GenericMapData; import org.apache.flink.cdc.common.data.MapData; import org.apache.flink.cdc.common.data.StringData; +import org.apache.flink.cdc.common.data.TimeData; import org.apache.flink.cdc.common.data.binary.BinaryArrayData; import org.apache.flink.cdc.common.data.binary.BinaryStringData; import org.apache.flink.cdc.common.types.DataTypes; @@ -162,4 +163,25 @@ void testToBinaryArrayWithDeeplyNestedTypes() { assertThat(arrayData2.getInt(0)).isEqualTo(44); assertThat(arrayData2.getInt(1)).isEqualTo(45); } + + @Test + void preservesHighPrecisionTimeInArraysAndMaps() { + long nanos = 3_723_123_456_789L; + ArrayDataSerializer arraySerializer = new ArrayDataSerializer(DataTypes.TIME(9)); + BinaryArrayData times = + arraySerializer.toBinaryArray( + new GenericArrayData(new Object[] {TimeData.fromNanoOfDay(nanos), null})); + + assertThat(times.getTime(0, 9).toNanoOfDay()).isEqualTo(nanos); + assertThat(times.isNullAt(1)).isTrue(); + + Map source = new HashMap<>(); + source.put(BinaryStringData.fromString("precise"), TimeData.fromNanoOfDay(nanos)); + MapDataSerializer mapSerializer = + new MapDataSerializer(DataTypes.STRING(), DataTypes.TIME(9)); + MapData map = mapSerializer.toBinaryMap(new GenericMapData(source)); + int valueIndex = map.keyArray().getString(0).toString().equals("precise") ? 0 : -1; + assertThat(valueIndex).isZero(); + assertThat(map.valueArray().getTime(valueIndex, 9).toNanoOfDay()).isEqualTo(nanos); + } } diff --git a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/serializer/data/TimeDataSerializerTest.java b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/serializer/data/TimeDataSerializerTest.java index 692ab444468..369d7040c54 100644 --- a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/serializer/data/TimeDataSerializerTest.java +++ b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/serializer/data/TimeDataSerializerTest.java @@ -18,21 +18,33 @@ package org.apache.flink.cdc.runtime.serializer.data; import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.api.common.typeutils.TypeSerializerSchemaCompatibility; +import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot; +import org.apache.flink.api.common.typeutils.TypeSerializerSnapshotSerializationUtil; import org.apache.flink.cdc.common.data.TimeData; import org.apache.flink.cdc.runtime.serializer.SerializerTestBase; +import org.apache.flink.core.memory.DataInputDeserializer; +import org.apache.flink.core.memory.DataOutputSerializer; +import org.junit.jupiter.api.Test; + +import java.io.ByteArrayInputStream; +import java.io.ObjectInputStream; import java.time.LocalTime; +import java.util.Base64; + +import static org.assertj.core.api.Assertions.assertThat; -/** A test for the {@link TimestampDataSerializer}. */ -class TimeDataSerializerTest extends SerializerTestBase { +/** Tests for {@link TimeDataSerializer}. */ +abstract class TimeDataSerializerTest extends SerializerTestBase { @Override protected TypeSerializer createSerializer() { - return TimeDataSerializer.INSTANCE; + return new TimeDataSerializer(getPrecision()); } @Override protected int getLength() { - return 4; + return getPrecision() <= 3 ? Integer.BYTES : Long.BYTES; } @Override @@ -42,21 +54,164 @@ protected Class getTypeClass() { @Override protected TimeData[] getTestData() { + if (getPrecision() <= 3) { + return new TimeData[] { + TimeData.fromSecondOfDay(1024), + TimeData.fromMillisOfDay(20480), + TimeData.fromIsoLocalTimeString("14:28:25.123"), + TimeData.fromLocalTime(LocalTime.NOON) + }; + } return new TimeData[] { - TimeData.fromSecondOfDay(1024), - TimeData.fromSecondOfDay(2048), - TimeData.fromSecondOfDay(4096), - TimeData.fromMillisOfDay(10240), - TimeData.fromMillisOfDay(20480), - TimeData.fromMillisOfDay(40960), - TimeData.fromNanoOfDay(102400), - TimeData.fromNanoOfDay(204800), - TimeData.fromNanoOfDay(409600), - TimeData.fromIsoLocalTimeString("14:28:25"), - TimeData.fromIsoLocalTimeString("01:23:45"), - TimeData.fromIsoLocalTimeString("23:59:59"), - TimeData.fromLocalTime(LocalTime.MIDNIGHT), - TimeData.fromLocalTime(LocalTime.NOON) + TimeData.fromNanoOfDay(102_400), + TimeData.fromNanoOfDay(20_480_123_456L), + TimeData.fromIsoLocalTimeString("14:28:25.123456789"), + TimeData.fromLocalTime(LocalTime.of(23, 59, 59, 999_999_999)) }; } + + protected abstract int getPrecision(); + + @Test + void roundTripRetainsDirectNumericPrecision() throws Exception { + long nanos = getPrecision() <= 3 ? 3_723_123_000_000L : 3_723_123_456_789L; + TimeDataSerializer serializer = new TimeDataSerializer(getPrecision()); + DataOutputSerializer output = new DataOutputSerializer(serializer.getLength()); + serializer.serialize(TimeData.fromNanoOfDay(nanos), output); + + TimeData restored = + serializer.deserialize(new DataInputDeserializer(output.getCopyOfBuffer())); + assertThat(restored.toNanoOfDay()).isEqualTo(nanos); + } +} + +final class TimeDataSerializer3Test extends TimeDataSerializerTest { + @Override + protected int getPrecision() { + return 3; + } +} + +final class TimeDataSerializer6Test extends TimeDataSerializerTest { + @Override + protected int getPrecision() { + return 6; + } +} + +final class TimeDataSerializer9Test extends TimeDataSerializerTest { + @Override + protected int getPrecision() { + return 9; + } +} + +final class TimeDataSerializerCompatibilityTest { + + private static final String LEGACY_SERIALIZER_BASE64 = + "rO0ABXNyAD9vcmcuYXBhY2hlLmZsaW5rLmNkYy5ydW50aW1lLnNlcmlhbGl6ZXIuZGF0YS5UaW1lRGF0YVNlcmlhbGl6ZXIAAAAAAAAAAQIAAHhyAD9vcmcuYXBhY2hlLmZsaW5rLmNkYy5ydW50aW1lLnNlcmlhbGl6ZXIuVHlwZVNlcmlhbGl6ZXJTaW5nbGV0b255qYeqxy53RQIAAHhyADRvcmcuYXBhY2hlLmZsaW5rLmFwaS5jb21tb24udHlwZXV0aWxzLlR5cGVTZXJpYWxpemVyAAAAAAAAAAECAAB4cA=="; + + private static final String LEGACY_SNAPSHOT_BASE64 = + "AAAAAgBab3JnLmFwYWNoZS5mbGluay5jZGMucnVudGltZS5zZXJpYWxpemVyLmRhdGEuVGltZURhdGFTZXJpYWxpemVyJFRpbWVEYXRhU2VyaWFsaXplclNuYXBzaG90AAAAAw=="; + + @Test + void oldMillisecondSnapshotIsCompatibleOrMigratableByPrecision() throws Exception { + TimeDataSerializer.TimeDataSerializerSnapshot oldSnapshot = + new TimeDataSerializer.TimeDataSerializerSnapshot(); + oldSnapshot.readSnapshot( + 3, + new DataInputDeserializer(new byte[0]), + Thread.currentThread().getContextClassLoader()); + + TypeSerializerSchemaCompatibility millisCompatibility = + oldSnapshot.resolveSchemaCompatibility(new TimeDataSerializer(3)); + TypeSerializerSchemaCompatibility microsCompatibility = + oldSnapshot.resolveSchemaCompatibility(new TimeDataSerializer(6)); + assertThat(millisCompatibility.isCompatibleAsIs()).isTrue(); + assertThat(microsCompatibility.isCompatibleAfterMigration()).isTrue(); + + DataOutputSerializer oldBytes = new DataOutputSerializer(Integer.BYTES); + oldBytes.writeInt(3_723_123); + TimeData restored = + oldSnapshot + .restoreSerializer() + .deserialize(new DataInputDeserializer(oldBytes.getCopyOfBuffer())); + assertThat(restored.toNanoOfDay()).isEqualTo(3_723_123_000_000L); + } + + @Test + void serializerCompatibilityDependsOnBinaryEncodingWidth() { + TimeDataSerializer.TimeDataSerializerSnapshot millisSnapshot = + (TimeDataSerializer.TimeDataSerializerSnapshot) + new TimeDataSerializer(3).snapshotConfiguration(); + TimeDataSerializer.TimeDataSerializerSnapshot nanosSnapshot = + (TimeDataSerializer.TimeDataSerializerSnapshot) + new TimeDataSerializer(6).snapshotConfiguration(); + + assertThat( + millisSnapshot + .resolveSchemaCompatibility(new TimeDataSerializer(0)) + .isCompatibleAsIs()) + .isTrue(); + assertThat( + nanosSnapshot + .resolveSchemaCompatibility(new TimeDataSerializer(9)) + .isCompatibleAsIs()) + .isTrue(); + assertThat( + millisSnapshot + .resolveSchemaCompatibility(new TimeDataSerializer(6)) + .isCompatibleAfterMigration()) + .isTrue(); + assertThat( + nanosSnapshot + .resolveSchemaCompatibility(new TimeDataSerializer(3)) + .isCompatibleAfterMigration()) + .isTrue(); + } + + @Test + void readsLegacySnapshotEnvelopeAndFourBytePayload() throws Exception { + TypeSerializerSnapshot snapshot = + TypeSerializerSnapshotSerializationUtil.readSerializerSnapshot( + new DataInputDeserializer( + Base64.getDecoder().decode(LEGACY_SNAPSHOT_BASE64)), + Thread.currentThread().getContextClassLoader()); + assertThat(snapshot).isInstanceOf(TimeDataSerializer.TimeDataSerializerSnapshot.class); + TimeDataSerializer.TimeDataSerializerSnapshot legacySnapshot = + (TimeDataSerializer.TimeDataSerializerSnapshot) snapshot; + + assertThat( + legacySnapshot + .resolveSchemaCompatibility(new TimeDataSerializer(3)) + .isCompatibleAsIs()) + .isTrue(); + DataOutputSerializer oldBytes = new DataOutputSerializer(Integer.BYTES); + oldBytes.writeInt(3_723_123); + assertThat( + snapshot.restoreSerializer() + .deserialize(new DataInputDeserializer(oldBytes.getCopyOfBuffer())) + .toNanoOfDay()) + .isEqualTo(3_723_123_000_000L); + } + + @Test + void readsLegacyJavaSerializedSingleton() throws Exception { + TimeDataSerializer serializer; + try (ObjectInputStream input = + new ObjectInputStream( + new ByteArrayInputStream( + Base64.getDecoder().decode(LEGACY_SERIALIZER_BASE64)))) { + serializer = (TimeDataSerializer) input.readObject(); + } + + assertThat(serializer.getLength()).isEqualTo(Integer.BYTES); + DataOutputSerializer oldBytes = new DataOutputSerializer(Integer.BYTES); + oldBytes.writeInt(3_723_123); + assertThat( + serializer + .deserialize(new DataInputDeserializer(oldBytes.getCopyOfBuffer())) + .toNanoOfDay()) + .isEqualTo(3_723_123_000_000L); + } } diff --git a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/typeutils/BinaryRecordDataExtractorTest.java b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/typeutils/BinaryRecordDataExtractorTest.java index e73f38fa1f1..8453ba47f73 100644 --- a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/typeutils/BinaryRecordDataExtractorTest.java +++ b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/typeutils/BinaryRecordDataExtractorTest.java @@ -174,8 +174,8 @@ void testConvertingBinaryRecordData() { Assertions.assertThat(generateEventWithAllTypes()) .map(e -> BinaryRecordDataExtractor.extractRecord(e, SCHEMA.toRowDataType())) .containsExactly( - "{id: INT NOT NULL -> 1, bool_col: BOOLEAN -> true, tinyint_col: TINYINT -> 2, smallint_col: SMALLINT -> 3, int_col: INT -> 4, bigint_col: BIGINT -> 5, float_col: FLOAT -> 6.0, double_col: DOUBLE -> 7.0, decimal_col: DECIMAL(17, 10) -> 0.1234567890, char_col: CHAR(17) -> Eight, varchar_col: VARCHAR(17) -> Nine, bin_col: BINARY(17) -> VGVuAQ==, varbin_col: VARBINARY(17) -> RWxldmVuAg==, date_col: DATE -> 2019-12-31, time_col: TIME(9) -> 08:30:17.123, ts_col: TIMESTAMP(3) -> 2023-11-11T11:11:11.000000011, ts_tz_col: TIMESTAMP(3) WITH TIME ZONE -> 2023-11-11T11:11:11.000000011+05:00, ts_ltz_col: TIMESTAMP_LTZ(3) -> 2023-11-11T06:11:11.000000011, array_col: ARRAY -> [One, Two, Three], map_col: MAP -> {1 -> yi, 2 -> er, 3 -> san}, row_col: ROW<`f0` INT, `f1` DOUBLE> -> {f0: INT -> 3, f1: DOUBLE -> 0.1415926}}", - "{id: INT NOT NULL -> -1, bool_col: BOOLEAN -> false, tinyint_col: TINYINT -> -2, smallint_col: SMALLINT -> -3, int_col: INT -> -4, bigint_col: BIGINT -> -5, float_col: FLOAT -> -6.0, double_col: DOUBLE -> -7.0, decimal_col: DECIMAL(17, 10) -> -0.1234567890, char_col: CHAR(17) -> -Eight, varchar_col: VARCHAR(17) -> -Nine, bin_col: BINARY(17) -> LVRlbgE=, varbin_col: VARBINARY(17) -> LUVsZXZlbgI=, date_col: DATE -> 2019-12-31, time_col: TIME(9) -> 08:30:17.123, ts_col: TIMESTAMP(3) -> 2021-11-11T11:11:11.000000011, ts_tz_col: TIMESTAMP(3) WITH TIME ZONE -> 2021-11-11T11:11:11.000000011+05:00, ts_ltz_col: TIMESTAMP_LTZ(3) -> 2021-11-11T06:11:11.000000011, array_col: ARRAY -> [Ninety, Eighty, Seventy], map_col: MAP -> {7 -> qi, 8 -> ba, 9 -> jiu}, row_col: ROW<`f0` INT, `f1` DOUBLE> -> {f0: INT -> 2, f1: DOUBLE -> 0.718281828}}", + "{id: INT NOT NULL -> 1, bool_col: BOOLEAN -> true, tinyint_col: TINYINT -> 2, smallint_col: SMALLINT -> 3, int_col: INT -> 4, bigint_col: BIGINT -> 5, float_col: FLOAT -> 6.0, double_col: DOUBLE -> 7.0, decimal_col: DECIMAL(17, 10) -> 0.1234567890, char_col: CHAR(17) -> Eight, varchar_col: VARCHAR(17) -> Nine, bin_col: BINARY(17) -> VGVuAQ==, varbin_col: VARBINARY(17) -> RWxldmVuAg==, date_col: DATE -> 2019-12-31, time_col: TIME(9) -> 08:30:17.123456789, ts_col: TIMESTAMP(3) -> 2023-11-11T11:11:11.000000011, ts_tz_col: TIMESTAMP(3) WITH TIME ZONE -> 2023-11-11T11:11:11.000000011+05:00, ts_ltz_col: TIMESTAMP_LTZ(3) -> 2023-11-11T06:11:11.000000011, array_col: ARRAY -> [One, Two, Three], map_col: MAP -> {1 -> yi, 2 -> er, 3 -> san}, row_col: ROW<`f0` INT, `f1` DOUBLE> -> {f0: INT -> 3, f1: DOUBLE -> 0.1415926}}", + "{id: INT NOT NULL -> -1, bool_col: BOOLEAN -> false, tinyint_col: TINYINT -> -2, smallint_col: SMALLINT -> -3, int_col: INT -> -4, bigint_col: BIGINT -> -5, float_col: FLOAT -> -6.0, double_col: DOUBLE -> -7.0, decimal_col: DECIMAL(17, 10) -> -0.1234567890, char_col: CHAR(17) -> -Eight, varchar_col: VARCHAR(17) -> -Nine, bin_col: BINARY(17) -> LVRlbgE=, varbin_col: VARBINARY(17) -> LUVsZXZlbgI=, date_col: DATE -> 2019-12-31, time_col: TIME(9) -> 08:30:17.123456789, ts_col: TIMESTAMP(3) -> 2021-11-11T11:11:11.000000011, ts_tz_col: TIMESTAMP(3) WITH TIME ZONE -> 2021-11-11T11:11:11.000000011+05:00, ts_ltz_col: TIMESTAMP_LTZ(3) -> 2021-11-11T06:11:11.000000011, array_col: ARRAY -> [Ninety, Eighty, Seventy], map_col: MAP -> {7 -> qi, 8 -> ba, 9 -> jiu}, row_col: ROW<`f0` INT, `f1` DOUBLE> -> {f0: INT -> 2, f1: DOUBLE -> 0.718281828}}", "{id: INT NOT NULL -> 0, bool_col: BOOLEAN -> null, tinyint_col: TINYINT -> null, smallint_col: SMALLINT -> null, int_col: INT -> null, bigint_col: BIGINT -> null, float_col: FLOAT -> null, double_col: DOUBLE -> null, decimal_col: DECIMAL(17, 10) -> null, char_col: CHAR(17) -> null, varchar_col: VARCHAR(17) -> null, bin_col: BINARY(17) -> null, varbin_col: VARBINARY(17) -> null, date_col: DATE -> null, time_col: TIME(9) -> null, ts_col: TIMESTAMP(3) -> null, ts_tz_col: TIMESTAMP(3) WITH TIME ZONE -> null, ts_ltz_col: TIMESTAMP_LTZ(3) -> null, array_col: ARRAY -> null, map_col: MAP -> null, row_col: ROW<`f0` INT, `f1` DOUBLE> -> null}", "null"); } @@ -185,8 +185,8 @@ void testConvertingBinaryRecordDataWithSchema() { Assertions.assertThat(generateEventWithAllTypes()) .map(e -> BinaryRecordDataExtractor.extractRecord(e, SCHEMA)) .containsExactly( - "{id: INT NOT NULL -> 1, bool_col: BOOLEAN -> true, tinyint_col: TINYINT -> 2, smallint_col: SMALLINT -> 3, int_col: INT -> 4, bigint_col: BIGINT -> 5, float_col: FLOAT -> 6.0, double_col: DOUBLE -> 7.0, decimal_col: DECIMAL(17, 10) -> 0.1234567890, char_col: CHAR(17) -> Eight, varchar_col: VARCHAR(17) -> Nine, bin_col: BINARY(17) -> VGVuAQ==, varbin_col: VARBINARY(17) -> RWxldmVuAg==, date_col: DATE -> 2019-12-31, time_col: TIME(9) -> 08:30:17.123, ts_col: TIMESTAMP(3) -> 2023-11-11T11:11:11.000000011, ts_tz_col: TIMESTAMP(3) WITH TIME ZONE -> 2023-11-11T11:11:11.000000011+05:00, ts_ltz_col: TIMESTAMP_LTZ(3) -> 2023-11-11T06:11:11.000000011, array_col: ARRAY -> [One, Two, Three], map_col: MAP -> {1 -> yi, 2 -> er, 3 -> san}, row_col: ROW<`f0` INT, `f1` DOUBLE> -> {f0: INT -> 3, f1: DOUBLE -> 0.1415926}}", - "{id: INT NOT NULL -> -1, bool_col: BOOLEAN -> false, tinyint_col: TINYINT -> -2, smallint_col: SMALLINT -> -3, int_col: INT -> -4, bigint_col: BIGINT -> -5, float_col: FLOAT -> -6.0, double_col: DOUBLE -> -7.0, decimal_col: DECIMAL(17, 10) -> -0.1234567890, char_col: CHAR(17) -> -Eight, varchar_col: VARCHAR(17) -> -Nine, bin_col: BINARY(17) -> LVRlbgE=, varbin_col: VARBINARY(17) -> LUVsZXZlbgI=, date_col: DATE -> 2019-12-31, time_col: TIME(9) -> 08:30:17.123, ts_col: TIMESTAMP(3) -> 2021-11-11T11:11:11.000000011, ts_tz_col: TIMESTAMP(3) WITH TIME ZONE -> 2021-11-11T11:11:11.000000011+05:00, ts_ltz_col: TIMESTAMP_LTZ(3) -> 2021-11-11T06:11:11.000000011, array_col: ARRAY -> [Ninety, Eighty, Seventy], map_col: MAP -> {7 -> qi, 8 -> ba, 9 -> jiu}, row_col: ROW<`f0` INT, `f1` DOUBLE> -> {f0: INT -> 2, f1: DOUBLE -> 0.718281828}}", + "{id: INT NOT NULL -> 1, bool_col: BOOLEAN -> true, tinyint_col: TINYINT -> 2, smallint_col: SMALLINT -> 3, int_col: INT -> 4, bigint_col: BIGINT -> 5, float_col: FLOAT -> 6.0, double_col: DOUBLE -> 7.0, decimal_col: DECIMAL(17, 10) -> 0.1234567890, char_col: CHAR(17) -> Eight, varchar_col: VARCHAR(17) -> Nine, bin_col: BINARY(17) -> VGVuAQ==, varbin_col: VARBINARY(17) -> RWxldmVuAg==, date_col: DATE -> 2019-12-31, time_col: TIME(9) -> 08:30:17.123456789, ts_col: TIMESTAMP(3) -> 2023-11-11T11:11:11.000000011, ts_tz_col: TIMESTAMP(3) WITH TIME ZONE -> 2023-11-11T11:11:11.000000011+05:00, ts_ltz_col: TIMESTAMP_LTZ(3) -> 2023-11-11T06:11:11.000000011, array_col: ARRAY -> [One, Two, Three], map_col: MAP -> {1 -> yi, 2 -> er, 3 -> san}, row_col: ROW<`f0` INT, `f1` DOUBLE> -> {f0: INT -> 3, f1: DOUBLE -> 0.1415926}}", + "{id: INT NOT NULL -> -1, bool_col: BOOLEAN -> false, tinyint_col: TINYINT -> -2, smallint_col: SMALLINT -> -3, int_col: INT -> -4, bigint_col: BIGINT -> -5, float_col: FLOAT -> -6.0, double_col: DOUBLE -> -7.0, decimal_col: DECIMAL(17, 10) -> -0.1234567890, char_col: CHAR(17) -> -Eight, varchar_col: VARCHAR(17) -> -Nine, bin_col: BINARY(17) -> LVRlbgE=, varbin_col: VARBINARY(17) -> LUVsZXZlbgI=, date_col: DATE -> 2019-12-31, time_col: TIME(9) -> 08:30:17.123456789, ts_col: TIMESTAMP(3) -> 2021-11-11T11:11:11.000000011, ts_tz_col: TIMESTAMP(3) WITH TIME ZONE -> 2021-11-11T11:11:11.000000011+05:00, ts_ltz_col: TIMESTAMP_LTZ(3) -> 2021-11-11T06:11:11.000000011, array_col: ARRAY -> [Ninety, Eighty, Seventy], map_col: MAP -> {7 -> qi, 8 -> ba, 9 -> jiu}, row_col: ROW<`f0` INT, `f1` DOUBLE> -> {f0: INT -> 2, f1: DOUBLE -> 0.718281828}}", "{id: INT NOT NULL -> 0, bool_col: BOOLEAN -> null, tinyint_col: TINYINT -> null, smallint_col: SMALLINT -> null, int_col: INT -> null, bigint_col: BIGINT -> null, float_col: FLOAT -> null, double_col: DOUBLE -> null, decimal_col: DECIMAL(17, 10) -> null, char_col: CHAR(17) -> null, varchar_col: VARCHAR(17) -> null, bin_col: BINARY(17) -> null, varbin_col: VARBINARY(17) -> null, date_col: DATE -> null, time_col: TIME(9) -> null, ts_col: TIMESTAMP(3) -> null, ts_tz_col: TIMESTAMP(3) WITH TIME ZONE -> null, ts_ltz_col: TIMESTAMP_LTZ(3) -> null, array_col: ARRAY -> null, map_col: MAP -> null, row_col: ROW<`f0` INT, `f1` DOUBLE> -> null}", "null"); } diff --git a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/typeutils/BinaryRecordDataGeneratorTest.java b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/typeutils/BinaryRecordDataGeneratorTest.java index d6c0dd4d647..b9cf157512f 100644 --- a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/typeutils/BinaryRecordDataGeneratorTest.java +++ b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/typeutils/BinaryRecordDataGeneratorTest.java @@ -32,6 +32,7 @@ import org.apache.flink.cdc.common.types.ZonedTimestampType; import org.apache.flink.cdc.common.types.variant.BinaryVariantBuilder; import org.apache.flink.cdc.common.types.variant.Variant; +import org.apache.flink.cdc.runtime.serializer.data.writer.BinaryRecordDataWriter; import org.junit.jupiter.api.Test; @@ -44,6 +45,31 @@ /** Unit tests for {@link BinaryRecordDataGenerator}. */ class BinaryRecordDataGeneratorTest { + @Test + void preservesHighPrecisionTimeAndReadsLegacyMilliseconds() { + BinaryRecordData precise = + new BinaryRecordDataGenerator( + RowType.of(DataTypes.TIME(3), DataTypes.TIME(6), DataTypes.TIME(9))) + .generate( + new Object[] { + TimeData.fromNanoOfDay(3_723_123_000_000L), + TimeData.fromNanoOfDay(3_723_123_456_000L), + TimeData.fromNanoOfDay(3_723_123_456_789L) + }); + + assertThat(precise.getTime(0, 3).toNanoOfDay()).isEqualTo(3_723_123_000_000L); + assertThat(precise.getTime(1, 6).toNanoOfDay()).isEqualTo(3_723_123_456_000L); + assertThat(precise.getTime(1, 0).toNanoOfDay()).isEqualTo(3_723_123_456_000L); + assertThat(precise.getTime(2, 9).toNanoOfDay()).isEqualTo(3_723_123_456_789L); + assertThat(precise.getTime(2).toNanoOfDay()).isEqualTo(3_723_123_456_789L); + + BinaryRecordData legacy = new BinaryRecordData(1); + BinaryRecordDataWriter writer = new BinaryRecordDataWriter(legacy); + writer.writeInt(0, 3_723_123); + writer.complete(); + assertThat(legacy.getTime(0, 6).toNanoOfDay()).isEqualTo(3_723_123_000_000L); + } + @Test void testOf() { RowType rowType = From 2e374785d3afe96198a19db991206515ff7d5a7e Mon Sep 17 00:00:00 2001 From: tchivs Date: Thu, 10 Sep 2026 15:52:34 +0800 Subject: [PATCH 2/3] [FLINK-40594][runtime&postgres] Fix the CI failures of the TIME precision change GenericRecordDataSerializer still wrote TIME as millisecond-of-day, so a GenericRecordData round trip truncated every TIME(p > 3). It now writes nanosecond-of-day under a new tag and keeps reading the legacy one. PostgresFullTypesITCase expected 18:00:22.123456 from the snapshot phase, but the snapshot reads TIME columns with ResultSet#getObject, which yields a java.sql.Time carrying only milliseconds; the truncation is upstream of the runtime representation. The snapshot expectation now records that, and microsecond retention is asserted on the change stream, where the emitted value really carries microseconds. TransformE2eITCase carried the same LOCALTIME expectation that was corrected in FlinkPipelineTransformITCase: LOCALTIME and CURRENT_TIME are TIME(0), so only whole seconds belong to their declared type. --- .../source/PostgresFullTypesITCase.java | 46 ++++++++++++++++--- .../pipeline/tests/TransformE2eITCase.java | 7 ++- .../data/GenericRecordDataSerializer.java | 10 +++- .../data/RecordDataSerializerTest.java | 14 ++++++ 4 files changed, 66 insertions(+), 11 deletions(-) diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/test/java/org/apache/flink/cdc/connectors/postgres/source/PostgresFullTypesITCase.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/test/java/org/apache/flink/cdc/connectors/postgres/source/PostgresFullTypesITCase.java index beb5eb4be96..01cecae8e0e 100644 --- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/test/java/org/apache/flink/cdc/connectors/postgres/source/PostgresFullTypesITCase.java +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/test/java/org/apache/flink/cdc/connectors/postgres/source/PostgresFullTypesITCase.java @@ -293,7 +293,11 @@ public void testTimeTypesWithTemporalModeAdaptive() throws Exception { DateData.fromEpochDay(18460), TimeData.fromLocalTime(LocalTime.parse("18:00:22")), TimeData.fromLocalTime(LocalTime.parse("18:00:22.123")), - TimeData.fromLocalTime(LocalTime.parse("18:00:22.123456")), + // The snapshot phase reads TIME columns through java.sql.Time, which only + // carries milliseconds, so time_6_c arrives truncated no matter how the + // runtime represents it. Microsecond retention is asserted on the change + // stream below. + TimeData.fromLocalTime(LocalTime.parse("18:00:22.123")), TimestampData.fromLocalDateTime(LocalDateTime.parse("2020-07-17T18:00:22")), TimestampData.fromLocalDateTime(LocalDateTime.parse("2020-07-17T18:00:22.123")), TimestampData.fromLocalDateTime( @@ -306,11 +310,8 @@ public void testTimeTypesWithTemporalModeAdaptive() throws Exception { RecordData snapshotRecord = ((DataChangeEvent) snapshotResults.get(0)).after(); Assertions.assertThat(recordFields(snapshotRecord, TIME_TYPES_WITH_ADAPTIVE)) .isEqualTo(expectedSnapshot); - // Without sub-millisecond retention the TIME(6) entry above and the value read back both - // truncate to 18:00:22.123, so the comparison passes while the microseconds are already - // gone. Assert the microsecond-of-day directly so the expectation cannot pass vacuously. - Assertions.assertThat(snapshotRecord.getTime(4, 6).toMicroOfDay()) - .isEqualTo(64_822_123_456L); + Assertions.assertThat(streamedMicroOfDayOfTime6Column(events, 3)) + .isEqualTo(68_422_123_456L); } @Test @@ -354,7 +355,9 @@ public void testTimeTypesWithTemporalModeMicroSeconds() throws Exception { DateData.fromEpochDay(18460), TimeData.fromLocalTime(LocalTime.parse("18:00:22")), TimeData.fromLocalTime(LocalTime.parse("18:00:22.123")), - TimeData.fromLocalTime(LocalTime.parse("18:00:22.123456")), + // Truncated by the java.sql.Time based snapshot read, see the change-stream + // assertion below. + TimeData.fromLocalTime(LocalTime.parse("18:00:22.123")), TimestampData.fromLocalDateTime(LocalDateTime.parse("2020-07-17T18:00:22")), TimestampData.fromLocalDateTime(LocalDateTime.parse("2020-07-17T18:00:22.123")), TimestampData.fromLocalDateTime( @@ -367,6 +370,8 @@ public void testTimeTypesWithTemporalModeMicroSeconds() throws Exception { RecordData snapshotRecord = ((DataChangeEvent) snapshotResults.get(0)).after(); Assertions.assertThat(recordFields(snapshotRecord, TIME_TYPES_WITH_ADAPTIVE)) .isEqualTo(expectedSnapshot); + Assertions.assertThat(streamedMicroOfDayOfTime6Column(events, 3)) + .isEqualTo(68_422_123_456L); } @Test @@ -1110,6 +1115,33 @@ private Tuple2, List> fetchResultsAndCreateTableEv return Tuple2.of(result, createTableEvents); } + /** + * Inserts a {@code time_types} row whose {@code time_6_c} value is {@code 19:00:22.123456} and + * returns the microsecond-of-day the pipeline emits for that column from the change stream. + * + *

The snapshot phase reads TIME columns with {@code ResultSet#getObject}, which yields a + * {@link java.sql.Time} carrying only milliseconds, so sub-millisecond retention can only be + * observed on the change stream. + */ + private long streamedMicroOfDayOfTime6Column(CloseableIterator events, int id) + throws Exception { + try (Connection connection = + PostgresTestBase.getJdbcConnection(POSTGIS_CONTAINER, "postgres"); + Statement statement = connection.createStatement()) { + statement.execute( + String.format( + "INSERT INTO inventory.time_types VALUES (%d, '2020-07-17'," + + " '19:00:22', '19:00:22.123', '19:00:22.123456'," + + " '2020-07-17 19:00:22', '2020-07-17 19:00:22.123'," + + " '2020-07-17 19:00:22.123456', '2020-07-17 19:00:22'," + + " '2020-07-17 19:00:22+08:00')", + id)); + } + RecordData streamRecord = + ((DataChangeEvent) fetchResultsAndCreateTableEvent(events, 1).f0.get(0)).after(); + return streamRecord.getTime(4, 6).toMicroOfDay(); + } + private Object[] recordFields(RecordData record, RowType rowType) { int fieldNum = record.getArity(); List fieldTypes = rowType.getChildren(); diff --git a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/TransformE2eITCase.java b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/TransformE2eITCase.java index 4f9c7bd19af..a223ef17b95 100644 --- a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/TransformE2eITCase.java +++ b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/TransformE2eITCase.java @@ -1435,10 +1435,13 @@ void verifyDataRecord(String recordLine) { .toInstant(ZoneOffset.UTC); long milliSecondsInOneDay = 24 * 60 * 60 * 1000; + // LOCALTIME and CURRENT_TIME are TIME(0), so only whole seconds are part of their declared + // type. Comparing against the millisecond-of-day of CURRENT_TIMESTAMP would require the + // runtime to carry a fraction the column type does not have. assertThat(TimeData.fromIsoLocalTimeString(localTime)) .isEqualTo( - TimeData.fromMillisOfDay( - (int) (instant.toEpochMilli() % milliSecondsInOneDay))); + TimeData.fromSecondOfDay( + (int) (instant.toEpochMilli() % milliSecondsInOneDay / 1000))); String localDate = tokens.get(5); assertThat(DateData.fromIsoLocalDateString(localDate)) diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/data/GenericRecordDataSerializer.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/data/GenericRecordDataSerializer.java index 4f5c406fb30..95af0c2853e 100644 --- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/data/GenericRecordDataSerializer.java +++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/data/GenericRecordDataSerializer.java @@ -66,12 +66,16 @@ public class GenericRecordDataSerializer { private static final byte TAG_ZONED_TIMESTAMP = 12; private static final byte TAG_LOCAL_ZONED_TIMESTAMP = 13; private static final byte TAG_DATE = 14; + + /** Legacy TIME encoding, millisecond-of-day; still read, no longer written. */ private static final byte TAG_TIME = 15; + private static final byte TAG_GENERIC_RECORD = 16; private static final byte TAG_BINARY_RECORD = 17; private static final byte TAG_ARRAY = 18; private static final byte TAG_MAP = 19; private static final byte TAG_VARIANT = 20; + private static final byte TAG_TIME_NANO = 21; private GenericRecordDataSerializer() {} @@ -177,8 +181,8 @@ static void serializeField(Object field, DataOutputView target) throws IOExcepti target.writeByte(TAG_DATE); target.writeInt(((DateData) field).toEpochDay()); } else if (field instanceof TimeData) { - target.writeByte(TAG_TIME); - target.writeInt(((TimeData) field).toMillisOfDay()); + target.writeByte(TAG_TIME_NANO); + target.writeLong(((TimeData) field).toNanoOfDay()); } else if (field instanceof GenericRecordData) { target.writeByte(TAG_GENERIC_RECORD); serialize((GenericRecordData) field, target); @@ -260,6 +264,8 @@ static Object deserializeField(DataInputView source) throws IOException { return DateData.fromEpochDay(source.readInt()); case TAG_TIME: return TimeData.fromMillisOfDay(source.readInt()); + case TAG_TIME_NANO: + return TimeData.fromNanoOfDay(source.readLong()); case TAG_GENERIC_RECORD: return deserialize(source); case TAG_BINARY_RECORD: diff --git a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/serializer/data/RecordDataSerializerTest.java b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/serializer/data/RecordDataSerializerTest.java index 15cbde474ec..473e819f493 100644 --- a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/serializer/data/RecordDataSerializerTest.java +++ b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/serializer/data/RecordDataSerializerTest.java @@ -130,6 +130,20 @@ void testGenericRecordDataWithVariousTypes() throws Exception { assertThat(deserialized.isNullAt(15)).isTrue(); } + @Test + void testGenericRecordDataKeepsSubMillisecondTime() throws Exception { + RecordDataSerializer serializer = RecordDataSerializer.INSTANCE; + GenericRecordData record = + GenericRecordData.of(TimeData.fromNanoOfDay(43_200_123_456_789L)); + + DataOutputSerializer out = new DataOutputSerializer(32); + serializer.serialize(record, out); + RecordData deserialized = + serializer.deserialize(new DataInputDeserializer(out.getCopyOfBuffer())); + + assertThat(deserialized.getTime(0).toNanoOfDay()).isEqualTo(43_200_123_456_789L); + } + @Test void testBinaryRecordDataWithVariousTypes() throws Exception { RecordDataSerializer serializer = RecordDataSerializer.INSTANCE; From f17875339bf251f7468a348e9f4d03ebec81fcca Mon Sep 17 00:00:00 2001 From: tchivs Date: Thu, 10 Sep 2026 21:00:26 +0800 Subject: [PATCH 3/3] [FLINK-40594][postgres] Point the snapshot TIME expectation at FLINK-39748 --- .../source/PostgresFullTypesITCase.java | 22 +++++++++++-------- 1 file changed, 13 insertions(+), 9 deletions(-) diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/test/java/org/apache/flink/cdc/connectors/postgres/source/PostgresFullTypesITCase.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/test/java/org/apache/flink/cdc/connectors/postgres/source/PostgresFullTypesITCase.java index 01cecae8e0e..179cee7def3 100644 --- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/test/java/org/apache/flink/cdc/connectors/postgres/source/PostgresFullTypesITCase.java +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/test/java/org/apache/flink/cdc/connectors/postgres/source/PostgresFullTypesITCase.java @@ -293,10 +293,13 @@ public void testTimeTypesWithTemporalModeAdaptive() throws Exception { DateData.fromEpochDay(18460), TimeData.fromLocalTime(LocalTime.parse("18:00:22")), TimeData.fromLocalTime(LocalTime.parse("18:00:22.123")), - // The snapshot phase reads TIME columns through java.sql.Time, which only - // carries milliseconds, so time_6_c arrives truncated no matter how the - // runtime represents it. Microsecond retention is asserted on the change - // stream below. + // PostgresScanFetchTask reads snapshot rows with a bare + // ResultSet#getObject, which yields a java.sql.Time carrying only + // milliseconds, so time_6_c arrives truncated no matter how the runtime + // represents it. PostgresConnection#getColumnValue already reads TIME as a + // string for exactly that reason, and FLINK-39748 routes the snapshot path + // through it; this expectation has to become 18:00:22.123456 again once that + // lands. Microsecond retention is asserted on the change stream below. TimeData.fromLocalTime(LocalTime.parse("18:00:22.123")), TimestampData.fromLocalDateTime(LocalDateTime.parse("2020-07-17T18:00:22")), TimestampData.fromLocalDateTime(LocalDateTime.parse("2020-07-17T18:00:22.123")), @@ -355,8 +358,8 @@ public void testTimeTypesWithTemporalModeMicroSeconds() throws Exception { DateData.fromEpochDay(18460), TimeData.fromLocalTime(LocalTime.parse("18:00:22")), TimeData.fromLocalTime(LocalTime.parse("18:00:22.123")), - // Truncated by the java.sql.Time based snapshot read, see the change-stream - // assertion below. + // Truncated by the java.sql.Time based snapshot read, as in the adaptive + // case above; see the change-stream assertion below. TimeData.fromLocalTime(LocalTime.parse("18:00:22.123")), TimestampData.fromLocalDateTime(LocalDateTime.parse("2020-07-17T18:00:22")), TimestampData.fromLocalDateTime(LocalDateTime.parse("2020-07-17T18:00:22.123")), @@ -1119,9 +1122,10 @@ private Tuple2, List> fetchResultsAndCreateTableEv * Inserts a {@code time_types} row whose {@code time_6_c} value is {@code 19:00:22.123456} and * returns the microsecond-of-day the pipeline emits for that column from the change stream. * - *

The snapshot phase reads TIME columns with {@code ResultSet#getObject}, which yields a - * {@link java.sql.Time} carrying only milliseconds, so sub-millisecond retention can only be - * observed on the change stream. + *

The snapshot phase reads TIME columns with a bare {@code ResultSet#getObject}, which + * yields a {@link java.sql.Time} carrying only milliseconds, so sub-millisecond retention can + * only be observed on the change stream until FLINK-39748 routes the snapshot path through + * {@code PostgresConnection#getColumnValue}. */ private long streamedMicroOfDayOfTime6Column(CloseableIterator events, int id) throws Exception {