Yaoxuan Wu created FLINK-41306:
----------------------------------
Summary: Parquet format writes TIMESTAMP values before 1970 with a
fractional second one second late (INT96, parquet.utc-timezone=false); Flink
cannot read them back
Key: FLINK-41306
URL: https://issues.apache.org/jira/browse/FLINK-41306
Project: Flink
Issue Type: Bug
Components: Formats (JSON, Avro, Parquet, ORC, SequenceFile)
Affects Versions: 2.2.1, 2.3.0, 2.0.1
Reporter: Yaoxuan Wu
{code:python}
import glob, os, tempfile
import pyarrow.parquet as pq
from pyflink.table import TableEnvironment, EnvironmentSettings
t_env = TableEnvironment.create(EnvironmentSettings.in_batch_mode())
t_env.get_config().set('pipeline.jars', os.environ['PARQUET_JARS']) #
flink-sql-parquet + hadoop client jars
t_env.get_config().set('parallelism.default', '1')
d = tempfile.mkdtemp()
t_env.execute_sql(f"CREATE TABLE p (id INT, ts TIMESTAMP(3)) "
f"WITH ('connector'='filesystem', 'path'='file://{d}',
'format'='parquet')")
t_env.execute_sql("INSERT INTO p VALUES (1, TIMESTAMP '1960-06-15
12:00:00.000'), "
"(2, TIMESTAMP '1960-06-15 12:00:00.250'), (3, TIMESTAMP
'1980-06-15 12:00:00.250')").wait()
for f in glob.glob(d + '/part-*'):
print(pq.read_table(f).to_pylist())
# run with TZ=UTC (JVM zone UTC; pyarrow reads INT96 as UTC)
# expected: ts = 1960-06-15 12:00:00, 1960-06-15 12:00:00.250, 1980-06-15
12:00:00.250
# actual: ts = 1960-06-15 12:00:00, 1960-06-15 12:00:01.250, 1980-06-15
12:00:00.250
t_env.execute_sql("SELECT * FROM p").print()
# fails: java.lang.IllegalArgumentException: nanos > 999999999 or < 0
{code}
With the default {{parquet.utc-timezone = false}}, a TIMESTAMP before
1970-01-01T00:00:00 (in the JVM zone) with a non-zero fraction of a second is
written one second late: pyarrow and Spark ({{spark.read.parquet}}, session
zone UTC) both read 12:00:01.25 for row 2. Flink's own reader fails on the same
file. Whole-second values and values after 1970 are correct;
{{parquet.utc-timezone = true}} round-trips correctly in Flink.
Cause: {{ParquetRowDataWriter.timestampToInt96}}, non-UTC branch:
{code:java}
Timestamp timestamp = timestampData.toTimestamp();
long mills = timestamp.getTime();
julianDay = (int) ((mills / MILLIS_IN_DAY) + JULIAN_EPOCH_OFFSET_DAYS);
nanosOfDay = ((mills % MILLIS_IN_DAY) / 1000) * NANOS_PER_SECOND +
timestamp.getNanos();
{code}
For {{mills < 0}} the remainder is negative and {{/ 1000}} truncates it toward
zero to whole seconds, while {{Timestamp.getNanos()}} is the fraction of the
*floored* second, so a negative remainder with a fraction gains one second
(mills = -1500: -1 s + 0.5 s = -0.5 s instead of -1.5 s). The resulting
nanos-of-day is also negative, which INT96 does not allow;
{{TimestampColumnReader.int96ToTimestamp}} then calls {{Timestamp.setNanos}}
with a negative value and throws. The UTC branch is linear in milliseconds and
stores the right instant. {{timestampToInt64}} has the same {{Timestamp}}-based
split in its non-UTC branch.
Reproduces on 2.0.1, 2.2.1, 2.3.0 and master.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
