[FLINK-39036][formats] Honor microsecond Avro timestamp logical types

[daguimu]

What is the purpose of the change

Fix FLINK-39036: both directions of the flink-avro RowData ⇄ GenericRecord conversion ignored the Avro timestamp-micros / local-timestamp-micros logical types and treated the long-encoded epoch value as milliseconds. Because the bug was symmetric, a Flink-internal round-trip happened to round-trip correctly while any cross-system round-trip drifted by a factor of 1000. The first revision of this PR addressed only the writer, which would have flipped that asymmetry the wrong way (writes correct, reads still wrong → internal round-trip broken). The current revision fixes both sides so reads and writes agree on the logical type.

Brief change log

• flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/RowDataToAvroConverters.java
  • In each TIMESTAMP_WITHOUT_TIME_ZONE and TIMESTAMP_WITH_LOCAL_TIME_ZONE converter (legacy and non-legacy mapping) check whether the target schema declares timestamp-micros / local-timestamp-micros. If yes, emit the value in microseconds; otherwise keep the existing millisecond output.
  • Add private helpers isMicrosLogicalType(Schema) and toEpochMicros(Instant).
• flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroToRowDataConverters.java
  • convertToTimestamp now takes the Flink type's precision and treats the incoming Long as microseconds when precision > 3, matching the precision mapping that AvroSchemaConverter already establishes between the Avro logical type and the Flink TIMESTAMP / TIMESTAMP_WITH_LOCAL_TIME_ZONE precision.
  • Negative epoch values are split with Math.floorDiv / Math.floorMod so the sub-millisecond component stays in the [0, 999_999] range that TimestampData.fromEpochMillis(long, int) requires.
• flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/RowDataToAvroConvertersTest.java (new)
  • Three writer test cases covering each updated branch under both micros and millis schemas.
• flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroToRowDataConvertersTest.java (new)
  • Reader cases for timestamp-micros, local-timestamp-micros, the millis regression case, and a pre-1970 negative micros case that exercises floor semantics.
  • Two end-to-end round-trip tests (writer + reader together) that assert a TimestampData round-trips through a timestamp-micros schema and a timestamp-millis schema.

Verifying this change

This change is covered by the new tests:

• RowDataToAvroConvertersTest#testTimestampWithLocalTimeZoneRespectsMicrosLogicalType — LocalZonedTimestampType(6) writer emits 1_704_164_645_123_456L under the micros schema and 1_704_164_645_123L under the millis schema.
• RowDataToAvroConvertersTest#testTimestampWithoutTimeZoneRespectsLocalMicrosLogicalType — TimestampType(6) writer emits the same numeric values for the local-timestamp-micros / -millis schemas.
• RowDataToAvroConvertersTest#testLegacyTimestampMappingRespectsMicrosLogicalType — legacyTimestampMapping=true path still honours timestamp-micros.
• AvroToRowDataConvertersTest#testTimestampMicrosLogicalTypeReadAsMicros and #testLocalTimestampMicrosLogicalTypeReadAsMicros — reader returns the correct sub-millisecond TimestampData for the two micros logical types.
• AvroToRowDataConvertersTest#testTimestampMillisPrecisionPreservesExistingBehavior — guards the existing millis behaviour for precision <= 3.
• AvroToRowDataConvertersTest#testNegativeMicrosTimestampHandlesFloorSemantics — pre-1970 micros value (-1500L) decodes to (-2 ms, 500_000 ns).
• AvroToRowDataConvertersTest#testRoundTripPreservesMicrosPrecision and #testRoundTripPreservesMillisPrecision — write-then-read round-trips of a TimestampData are no-ops under both timestamp-micros and timestamp-millis schemas.

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): no
• The serializers: no
• The runtime per-record code paths (performance sensitive): no (one schema-logical-type check on the writer side before the existing branch; a captured-isMicros boolean on the reader side)
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
• The S3 file system connector: no

Documentation

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

[flinkbot]

CI report:

• aae33cc Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[davidradl]

nit: this should be an import

[daguimu]

Good catch on cleaning up the FQN. Unfortunately org.apache.flink.table.types.logical.LogicalType is already imported on line 28 and used at ~8 sites in this file (e.g. createConverter(LogicalType type) on line 76, LogicalType[] fieldTypes on line 282, etc.), so an unqualified import org.apache.avro.LogicalType would collide. Using the FQN at this single helper-method call site looked less invasive than FQN-ing every Flink LogicalType reference.

Happy to flip it the other way (import the Avro one, FQN the Flink ones) if you'd prefer — let me know which you'd like to see.

[davidradl]

I agree the the idea here, but am concerned that introducing this behaviour might result in a regression for existing applications. I wonder if it would be safer to introduce the new behaviour under a config flag until there is a version change.

[daguimu]

Thanks for the careful read.

My read is that this is a correctness fix rather than a behavior change worth preserving. The read path in AvroToRowDataConverters already honors timestamp-micros / local-timestamp-micros and returns microseconds, while the write path here ignores the same logical type and always emits milliseconds. A pipeline that round-trips through Flink with a timestamp-micros schema silently scales timestamps by 1000× — that's a data-corruption bug, not a stable contract.

The fix is also gated by what the schema declares: users whose schemas are timestamp-millis (the previous behavior for both branches) see no change at all. Only users who already declared timestamp-micros and were silently getting wrong values are affected, and for them the new output is what their schema asked for.

That said, the scope question deserves a committer's call. Could one of the flink-avro maintainers weigh in on whether a config flag is warranted on master? If so, happy to add one as a follow-up.

[daguimu]

Apologies — I need to correct my earlier comment. I claimed AvroToRowDataConverters already honors timestamp-micros, but on closer reading it does not: convertToTimestamp treats every incoming Long as milliseconds regardless of the schema's logical type, and there's no Conversions.addLogicalTypeConversion registered anywhere in flink-avro.

That means today both read and write ignore the logical type symmetrically, which is why a Flink-internal round-trip with a timestamp-micros schema happens to round-trip correctly (two bugs cancel out). Landing only the write-side fix in this PR would break that internal round-trip — exactly the regression you flagged. Your concern was well-founded; I had it wrong.

I'll expand the scope of this PR to also fix the read path so the bug is addressed symmetrically. That removes the regression risk and makes a config flag unnecessary. Will push a follow-up commit and update the PR title accordingly.

[daguimu]

@davidradl summary of the new revision (force-pushed to aae33cc):

Scope change. Following up on the discussion above — the writer-only fix would have flipped a symmetric bug into an asymmetric one and broken Flink-internal round-trips with a timestamp-micros schema. To avoid that regression I expanded the PR to fix the reader as well:

• RowDataToAvroConverters — unchanged from the previous revision. Writer honors the schema's logical type when emitting the long value.
• AvroToRowDataConverters.convertToTimestamp — now takes the Flink type precision and decodes the long as microseconds when precision > 3. The mapping precision↔logical-type is the one AvroSchemaConverter already uses in both directions, so in normal Flink usage the reader and writer agree without needing to plumb the schema through the read API.
• Negative epoch micros are decoded with Math.floorDiv / Math.floorMod so the nano-of-millis component stays in the [0, 999_999] range that TimestampData.fromEpochMillis(long, int) requires.

Tests. A new AvroToRowDataConvertersTest covers the reader for timestamp-micros, local-timestamp-micros, the millis regression case, and a pre-1970 negative-micros case. There are also two end-to-end round-trip tests that drive both converters together — these are the guard against the read/write asymmetry that triggered this discussion in the first place.

PR title and description are updated to reflect the bidirectional scope. The Spotless and import-collision feedback from your earlier reviews are still addressed in the writer file. Happy to keep iterating if anything looks off.

[github-actions]

This PR is being marked as stale since it has not had any activity in the last 90 days.
If you would like to keep this PR alive, please leave a comment asking for a review.
If the PR has merge conflicts, update it with the latest from the base branch.

If you are having difficulty finding a reviewer, please reach out to the
community, contact details can be found here: https://flink.apache.org/what-is-flink/community/

If this PR is no longer valid or desired, please feel free to close it.
If no activity occurs in the next 30 days, it will be automatically closed.