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

[tchivs]

What is the purpose of this pull request?

Fixes FLINK-40594: the pipeline runtime stores TIME as millisecond-of-day, so every TIME(p) with p > 3 is silently truncated even though TimeType accepts a precision of up to 9 and DebeziumEventDeserializationSchema already builds values through fromMicroOfDay / fromNanoOfDay.

Beyond the lost digits, TimeData.equals, hashCode and compareTo compared millisOfDay, so two TIME(6) values differing only in microseconds were indistinguishable and collapsed during deduplication, ordering and key comparison.

Brief change log

• TimeData stores nanosecond-of-day and exposes toMicroOfDay() / toNanoOfDay(); equality, hash and comparison now use that value. toMillisOfDay() keeps its signature and truncating behaviour, so existing callers stay source and binary compatible.
• TimeDataSerializer becomes precision-aware, following the existing shape of TimestampDataSerializer / LocalZonedTimestampDataSerializer instead of introducing a new pattern:
  • precision <= 3 keeps the historical four-byte millisecond encoding and resolves as isCompatibleAsIs, so existing state keeps working;
  • 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 readers/writers (BinaryRecordData, BinaryArrayData, AbstractBinaryWriter, BinaryArrayWriter), the converters and InternalSerializers thread the declared precision through.
• Two existing expectations that pinned the truncation are corrected rather than relaxed:
  • the TIME(6) / TIME(9) transform expectation now asserts the retained fraction instead of 21:48:25.123;
  • the temporal-function check compares LOCALTIME at whole-second granularity, because LOCALTIME and CURRENT_TIME are TIME(0) and previously carried a fraction their declared type does not have.

---

Verifying this change

This change added tests and can be verified as follows:

• Added unit tests in flink-cdc-common/src/test/java/org/apache/flink/cdc/common/data/TimeDataTest.java covering nanosecond retention and the equality/ordering behaviour.
• Updated unit tests in flink-cdc-runtime/.../serializer/data/TimeDataSerializerTest.java: it becomes an abstract base with concrete TimeDataSerializer3Test / 6Test / 9Test, plus a TimeDataSerializerCompatibilityTest asserting isCompatibleAsIs for the millisecond encoding, isCompatibleAfterMigration for the widened one, and successful reading of the legacy snapshot envelope and four-byte payload.
• Updated unit tests in flink-cdc-common/.../converter/InternalObjectConverterTest.java and JavaObjectConverterTest.java.
• Updated integration tests in flink-cdc-composer/.../FlinkPipelineTransformITCase.java and flink-cdc-pipeline-connector-postgres/.../PostgresFullTypesITCase.java. The Postgres case gains a direct microsecond-of-day assertion, because the existing TIME(6) expectation compared two equally truncated values and therefore passed vacuously.

Local results on top of 9f23c0356, built and run on the JDK 11 baseline (openjdk 11.0.24; verified the emitted bytecode is class file major 55):

Module	Result
flink-cdc-common	96 tests, 0 failures
flink-cdc-runtime	992 tests, 0 failures, 0 errors, 1 skipped
flink-cdc-composer (FlinkPipelineTransformITCase)	71 tests, 0 failures
Spotless	passes

The same modules were also green on JDK 17.

Two notes for reviewers:

• Existing TimeDataSerializerTest data included TimeData.fromNanoOfDay(102400), 204800 and 409600. Under millisecond storage all three collapsed to zero at construction, so the previous round-trip assertion passed while the intended distinct values never existed; the precision-parameterised tests replace it.
• The precision-parameterised serializer tests cover TIME(3) on the legacy four-byte encoding, TIME(6) and TIME(9) on the widened one, and the snapshot compatibility matrix, so a regression in either encoding path fails a specific test rather than a shared one.

Documentation

• Does this pull request introduce a new feature? (no)
• If yes, how is the feature documented? (not applicable)

---

Update: follow-up commit for the CI failures

GenericRecordDataSerializer still encoded TIME as millisecond-of-day, so a GenericRecordData round trip kept truncating TIME(p > 3). It now writes nanosecond-of-day under a new tag and still reads the legacy tag, covered by RecordDataSerializerTest#testGenericRecordDataKeepsSubMillisecondTime.

Behaviour note for the change log: because the internal converters normalise a value to its declared precision, LOCALTIME and CURRENT_TIME — both TIME(0) — now emit whole seconds instead of carrying a millisecond fraction their declared type does not have. FlinkPipelineTransformITCase and TransformE2eITCase assert that.

PostgresFullTypesITCase expected 18:00:22.123456 from the snapshot phase. That is not reachable today: PostgresScanFetchTask reads snapshot rows with ResultSet#getObject, and pgjdbc materialises time as java.sql.Time, which only carries milliseconds — the value is already truncated before it reaches the runtime. Measured on this branch: the fixture row time_6_c = 18:00:22.123456 arrives from the snapshot as nanoOfDay = 64822123000000 (18:00:22.123), while a row inserted afterwards with time_6_c = 19:00:22.123456 arrives from the change stream as nanoOfDay = 68422123456000 (19:00:22.123456). The snapshot expectation now records the millisecond resolution, and microsecond retention is asserted on the change stream. The snapshot read itself looks like a separate [postgres] defect and is left for its own JIRA.

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude Code (Anthropic)

The change was prepared, verified and described with AI assistance; I have reviewed every hunk, ran the suites listed above, and take responsibility for the correctness of the change.

[yuxiqian]

Thanks Lanny for the contribution, please check the CI failing cases as they seems relevant.

@leonardBang IIRC the TIME(3) precision has been discussed earlier, and the conclusion is we'd better keep the same memory layout with Flink SQL, truncating to 3-digits. As apache/flink#26954 has extended precision up to 9, should this be reconsidered now?

[leonardBang]

Thanks Lanny for the contribution, please check the CI failing cases as they seems relevant.

@leonardBang IIRC the TIME(3) precision has been discussed earlier, and the conclusion is we'd better keep the same memory layout with Flink SQL, truncating to 3-digits. As apache/flink#26954 has extended precision up to 9, should this be reconsidered now?

Thanks @tchivs for the contribution and @yuxiqian for the investigation, I think it's time to support TIME(p) with p > 3

[tchivs]

Thanks @yuxiqian and @leonardBang. I pushed 2e37478 which fixes the CI failures; all six red jobs came down to three causes, and only two of them were failures of this change.

1. TransformE2eITCase#testTemporalFunctions (the four Pipeline E2E jobs) — my oversight.
verifyDataRecord carried the same LOCALTIME expectation that I had already corrected in FlinkPipelineTransformITCase, and I missed the E2E copy. Since the converters normalise a value to its declared precision, and LOCALTIME / CURRENT_TIME are TIME(0), the pipeline now emits whole seconds there; both tests assert that consistently. I have added this to the change log in the description, because it is user-visible.

2. GenericRecordDataSerializer — an incompleteness of the fix that CI does not cover.
It still encoded TIME as millisecond-of-day, so a GenericRecordData round trip truncated TIME(p > 3) even after the binary path was fixed. It now writes nanosecond-of-day under a new tag and keeps reading the legacy tag, so old state stays readable. RecordDataSerializerTest#testGenericRecordDataKeepsSubMillisecondTime fails without that change.

3. PostgresFullTypesITCase#testTimeTypesWithTemporalMode{Adaptive,MicroSeconds} (Pipeline Unit Tests 1.x and 2.x) — a pre-existing PostgreSQL snapshot defect that this change made visible.
The 18:00:22.123456 expectation is not reachable in the snapshot phase: PostgresScanFetchTask reads snapshot rows with ResultSet#getObject, and pgjdbc materialises time as java.sql.Time, which only carries milliseconds, so the microseconds are gone before the value reaches the runtime. Under millisecond storage both sides of that assertion were equally truncated, which is why it used to pass. Measured on this branch:

phase	time_6_c in PostgreSQL	emitted nanoOfDay
snapshot	18:00:22.123456	64822123000000 (18:00:22.123)
change stream	19:00:22.123456	68422123456000 (19:00:22.123456)

So the runtime change does preserve microseconds where the source actually delivers them. The snapshot expectation now states the millisecond resolution with a comment, and microsecond retention is asserted on the change stream instead of the vacuous snapshot comparison. Fixing the snapshot read looks like a separate [postgres] change to me (reading Types.TIME as LocalTime rather than java.sql.Time); I would rather file that as its own JIRA than widen this PR — happy to do so if you agree.

On the memory-layout question: the serializer keeps the historical four-byte millisecond encoding and isCompatibleAsIs for precision <= 3, and only uses the eight-byte nanosecond-of-day encoding for precision > 3, reached through a versioned snapshot that still reads the legacy envelope. So TIME(3) state and layout are unchanged, and only the precisions that Flink SQL has now extended in apache/flink#26954 use the wider slot.

Verified locally on the JDK 11 target (built and run with JDK 17, Flink 1.20.3 profile): flink-cdc-common and flink-cdc-runtime 993 tests green, PostgresFullTypesITCase time-type and full-type cases green, FlinkPipelineTransformITCase#testDateAndTimeCastingFunctions and #testTransformWithTemporalFunction green, flink-cdc-pipeline-e2e-tests compiles, Spotless passes on the touched modules. I do not have a local environment to run the pipeline E2E suite itself, so I am relying on CI for that job.

[tchivs]

CI came back on 2e37478: 18 of 19 checks green, and every case that this PR actually broke now passes.

• Pipeline Unit Tests and Pipeline Connectors Unit Tests 2.x are green, so PostgresFullTypesITCase is fixed on both Flink lines.
• Pipeline E2E Tests (1-Parallelism), Pipeline E2E Tests 2.x (1-Parallelism) and Pipeline E2E Tests 2.x (4-Parallelism) are green.
• Only Pipeline E2E Tests (4-Parallelism) / test (11, 1.20.3, pipeline_e2e) is red, and TransformE2eITCase reports Tests run: 24, Failures: 0, Errors: 1 there — no assertion failure, so testTemporalFunctions passed in that job too.

The two remaining failures look unrelated to TIME:

1. TransformE2eITCase#testAssortedSchemaTransform(boolean)[1] — TimeoutException: failed to get specific event: CreateTableEvent{tableId=transform_test_duaf6z.terminus, schema=columns={ID INT NOT NULL, VERSION STRING, NAME STRING}, ...}. No temporal column is involved; the job simply did not reach that event inside the timeout.
2. MySqlToIcebergE2eITCase#testSyncWholeDatabase — the collected rows still had 10 fields while the expectation has 16 (..., 1, 1, 1, 1, 1, 1, 1, 1, 1, 1), i.e. the assertion ran before the added columns had propagated. Again no TIME column in the diff.

Both are in the 4-parallelism 1.20.3 job only, and that job flakes on master as well, independently of this PR: nightly run 34072345338 (2026-09-07) failed the same job on MySqlToHudiE2eITCase#testSyncWholeDatabase, and run 33935167454 (2026-09-05) failed the 1- and 4-parallelism E2E jobs.

I cannot re-run it myself (Must have admin rights to Repository). @yuxiqian could you kick off a re-run of that single job when you get a chance? If you would rather see a fresh full run, I can push a no-op commit instead — just say which you prefer.