kumarpritam863 opened a new pull request, #29521:
URL: https://github.com/apache/flink/pull/29521
## What
`AvroToRowDataConverters` and `RowDataToAvroConverters` now follow the unit
`AvroSchemaConverter` declares for a timestamp column: microseconds when the
mapping is not legacy and the precision is above 3, milliseconds otherwise.
## Why
For `TIMESTAMP(4..6)` and `TIMESTAMP_LTZ(4..6)` the schema says
`local-timestamp-micros` / `timestamp-micros`, but:
- the reader took every `long` as milliseconds, so a microsecond value was
read a thousand times too large;
- an `Instant` went through `toEpochMilli()`, which drops the
sub-millisecond digits;
- the writer wrote every timestamp as epoch milliseconds, also into a field
declared as microseconds.
Reading a long as microseconds with a truncating `/` and `%` would also fail
for a negative value that is not a whole millisecond (for example
`-86400000001`, 1969-12-30T23:59:59.999999Z): the nanosecond part comes out
negative and `TimestampData.fromEpochMillis` throws an
`IllegalArgumentException` without a message. This is how a job reading such a
record from Kafka ended up restarting forever; it was found by comparing a
Flink sink with a Kafka Connect sink on the same records.
## Change
- Reader: a `long` is split with `Math.floorDiv` / `Math.floorMod` when the
column is a microsecond one; an `Instant` goes through
`TimestampData.fromInstant` there; milliseconds columns and the legacy mapping
are untouched.
- Writer: a microsecond column is written as microseconds, computed from the
seconds and the (never negative) nanosecond part. `ChronoUnit.MICROS.between`
is not used because it counts nanoseconds on the way and overflows outside the
years 1678 to 2262.
- `TIME`: a `long` is accepted (a `time-micros` that the Avro reader gave
without a logical type conversion). `RowData` stores a TIME as an int of
milliseconds, so the microseconds are dropped there; this is stated in the code.
## Compatibility
Data written earlier with a `TIMESTAMP(4..6)` column under the non-legacy
mapping holds milliseconds in a field declared as microseconds, and is now read
as microseconds. The legacy mapping and precision 3 and below are unchanged.
`AvroRowDataDeSerializationSchemaTest#testTimestampTypeNewMapping` asserted
the old reading (123456 microseconds as `00:02:03.456`); it now expects
`00:00:00.123456`.
## Testing
- New `AvroToRowDataConvertersTest` (reader) and
`RowDataToAvroConvertersTest` (writer, and a write-then-read round trip),
covering 0, +-1, -999, -1000, -1001, -86400000000/-86400000001 (one day before
the epoch), 0001-01-01, the last microsecond of 9999 and a present-day value
with microsecond digits, for `TIMESTAMP` and `TIMESTAMP_LTZ`, plus the
unchanged millisecond, legacy and `Instant` cases.
- On `master` without the change, 36 of the 65 tests in these two classes
and the updated existing test fail or error; with it they pass.
- `mvn test` for `flink-avro` (444 tests, 57 skipped as before) and
`flink-avro-confluent-registry` (30 tests): 0 failures; Checkstyle 0
violations, Spotless clean.
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (please specify the tool below)
Generated-by: Claude Code (Anthropic Claude Sonnet 5.5)
---
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
