Pritam Kumar created FLINK-41026:
------------------------------------
Summary: [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
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)
