[
https://issues.apache.org/jira/browse/FLINK-40594?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18112627#comment-18112627
]
tchivs commented on FLINK-40594:
--------------------------------
I would like to take this one.
My understanding of the root cause: the type system and the source path already
carry sub-millisecond TIME, and only the pipeline runtime's storage narrows it.
{{TimeType}} accepts {{0..9}} and {{DebeziumEventDeserializationSchema}} builds
values through {{fromMicroOfDay}} / {{fromNanoOfDay}}, but {{TimeData}} holds
{{int millisOfDay}}, so the fraction is dropped at construction and cannot be
recovered downstream. The equality/ordering collapse follows from the same
field, which is why I treat this as a correctness issue rather than a rendering
one.
Design I would propose, keeping state compatibility intact:
* {{TimeData}} stores nanosecond-of-day and exposes {{toMicroOfDay()}} /
{{toNanoOfDay()}}; {{toMillisOfDay()}} keeps its signature and truncating
behaviour, so callers stay source and binary compatible.
* {{TimeDataSerializer}} becomes precision-aware, following the existing shape
of {{TimestampDataSerializer}} and {{LocalZonedTimestampDataSerializer}} rather
than inventing a new pattern:
** {{precision <= 3}} keeps the historical four-byte millisecond encoding and
resolves as {{isCompatibleAsIs}};
** {{precision > 3}} uses an eight-byte nanosecond-of-day encoding and resolves
as {{isCompatibleAfterMigration}}, through a versioned
{{TypeSerializerSnapshot}} that still reads the legacy snapshot envelope and
four-byte payload.
* The binary writers/readers and {{InternalSerializers}} thread the declared
precision through.
POC status: I have this implemented on top of master ({{9f23c0356}}) and
verified locally.
* {{flink-cdc-common}}: 46 tests pass, including a new {{TimeDataTest}}.
* {{flink-cdc-runtime}}: 992 tests pass, 0 failures. {{TimeDataSerializerTest}}
becomes an abstract base with concrete {{TimeDataSerializer3Test}} / {{6Test}}
/ {{9Test}}, plus a {{TimeDataSerializerCompatibilityTest}} that asserts
{{isCompatibleAsIs}} for the millisecond encoding,
{{isCompatibleAfterMigration}} for the widened one, and successful reading of
the legacy envelope and four-byte payload.
* {{flink-cdc-composer}}: {{FlinkPipelineTransformITCase}} passes. Two
expectations that pinned the truncation 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, since {{LOCALTIME}} and {{CURRENT_TIME}} are
{{TIME(0)}} and previously carried a fraction their declared type does not have.
* {{PostgresFullTypesITCase}} gains a direct microsecond-of-day assertion so
the existing {{TIME(6)}} expectation can no longer pass vacuously.
* Spotless passes.
Happy to adjust the direction if the community prefers a different
compatibility strategy. I will open the PR against master once this is assigned.
> Pipeline runtime silently truncates TIME(p > 3) to milliseconds
> ---------------------------------------------------------------
>
> Key: FLINK-40594
> URL: https://issues.apache.org/jira/browse/FLINK-40594
> Project: Flink
> Issue Type: Bug
> Components: Flink CDC
> Affects Versions: cdc-3.5.0, cdc-3.6.0
> Reporter: tchivs
> Priority: Major
>
> Flink CDC's pipeline runtime stores TIME as millisecond-of-day, while
> {{TimeType}} accepts a precision of up to 9 and the Debezium deserializer
> already produces microsecond and nanosecond values. Every {{TIME(p)}} with
> {{p > 3}} is therefore silently truncated, and two values that differ only
> below the millisecond become indistinguishable.
> h2. How to reproduce
> {code:java}
> LocalTime t = LocalTime.parse("04:05:06.123456");
> TimeData d = TimeData.fromLocalTime(t);
> System.out.println(d.toLocalTime()); // 04:05:06.123 -- expected
> 04:05:06.123456
> TimeData a = TimeData.fromLocalTime(LocalTime.parse("04:05:06.123456"));
> TimeData b = TimeData.fromLocalTime(LocalTime.parse("04:05:06.123999"));
> System.out.println(a.equals(b)); // true -- expected false
> {code}
> Executed against {{flink-cdc-common}} on master ({{9f23c0356}}). The same
> storage is present in the {{release-3.5.0}} and {{release-3.6.0}} tags:
> {{TimeData}} declares {{private final int millisOfDay}} and contains no
> {{toMicroOfDay}}, so both released lines are affected. {{TimeData}} was
> introduced with millisecond storage in FLINK-38079, so the behaviour dates
> from that change; {{release-3.4}} does not contain the class.
> h2. Where the precision is lost
> * {{TimeType.MAX_PRECISION = 9}} and the constructor accepts {{0..9}}, so the
> type system advertises a precision the data class cannot hold.
> * {{TimeData}} keeps {{private final int millisOfDay}}. {{fromMicroOfDay}}
> divides by 1000 and {{fromNanoOfDay}} divides by 1_000_000, discarding the
> sub-millisecond part at construction. There is no {{toMicroOfDay()}} /
> {{toNanoOfDay()}} accessor.
> * Sources already deliver sub-millisecond values:
> {{DebeziumEventDeserializationSchema}} calls {{TimeData.fromMicroOfDay}} for
> {{MicroTime}} and {{TimeData.fromNanoOfDay}} for {{NanoTime}}. The precision
> is parsed correctly and then dropped.
> * {{TimeDataSerializer}} is a precision-less singleton with a fixed four-byte
> encoding: {{getLength()}} returns 4, {{serialize}} writes
> {{writeInt(record.toMillisOfDay())}} and {{deserialize}} reads an int.
> {{copy(TimeData)}} round-trips through {{toMillisOfDay()}} and is lossy even
> in memory. {{BinaryRecordData}}, {{BinaryArrayData}} and
> {{GenericRecordDataSerializer}} read TIME as an int as well.
> * {{equals}}, {{hashCode}} and {{compareTo}} compare {{millisOfDay}}. Two
> {{TIME(6)}} values differing only in microseconds compare equal, so they
> collapse during deduplication, ordering and key comparison. This is a
> correctness problem, not only a rendering one.
> h2. Impact
> Any pipeline reading {{TIME(4..9)}} -- PostgreSQL {{time(6)}}, MySQL
> {{TIME(6)}}, and anything routed through the Debezium {{MicroTime}} /
> {{NanoTime}} semantic types -- loses sub-millisecond data, and TIME keys or
> deduplication treat distinct values as identical.
> h2. Two existing tests currently hide the defect
> * {{TimeDataSerializerTest.getTestData()}} contains
> {{TimeData.fromNanoOfDay(102400)}}, {{204800}} and {{409600}}. All three
> collapse to zero at construction, so the round-trip assertion passes while
> the intended distinct nanosecond values never existed.
> * {{PostgresFullTypesITCase}} expects
> {{TimeData.fromLocalTime(LocalTime.parse("18:00:22.123456"))}} for the
> {{TIME(6)}} column. The expected and the actual value both truncate to
> {{18:00:22.123}}, so the assertion passes vacuously.
> h2. Suggested direction
> Keep nanosecond-of-day inside {{TimeData}}, expose {{toMicroOfDay()}} /
> {{toNanoOfDay()}}, and compare on the nanosecond value. Make
> {{TimeDataSerializer}} precision-aware, as {{TimestampDataSerializer}} and
> {{LocalZonedTimestampDataSerializer}} already are: keep the four-byte
> millisecond encoding for {{precision <= 3}} so existing state stays
> compatible, and use an eight-byte nanosecond encoding above it behind a
> versioned {{TypeSerializerSnapshot}} that can still read the legacy snapshot
> envelope and payload. {{toMillisOfDay()}} would keep its signature and
> truncating behaviour, so existing callers stay source and binary compatible.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
