[
https://issues.apache.org/jira/browse/FLINK-41026?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-41026:
-----------------------------------
Labels: pull-request-available (was: )
> [format/avro] Convert timestamps above precision 3 as microseconds in both
> directions
> -------------------------------------------------------------------------------------
>
> Key: FLINK-41026
> URL: https://issues.apache.org/jira/browse/FLINK-41026
> Project: Flink
> Issue Type: Bug
> Components: Formats (JSON, Avro, Parquet, ORC, SequenceFile)
> Reporter: Pritam Kumar
> Assignee: Pritam Kumar
> Priority: Minor
> Labels: pull-request-available
>
> h2. 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.
> h2. 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.
> h2. 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.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
