tchivs created FLINK-40594:
------------------------------
Summary: 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.6.0, cdc-3.5.0
Reporter: tchivs
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)
