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..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,7 +293,14 @@ 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")), + // 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")), TimestampData.fromLocalDateTime( @@ -306,6 +313,8 @@ public void testTimeTypesWithTemporalModeAdaptive() 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 @@ -349,7 +358,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, 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")), TimestampData.fromLocalDateTime( @@ -362,6 +373,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 @@ -1105,6 +1118,34 @@ 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 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 { + 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/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/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/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/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; 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 =