[Bug] Parquet timestamp predicate pushdown returns incorrect results when file and read timestamp units differ

[lilei1128]

Search before asking

• I searched in the issues and found nothing similar.

Paimon version

paimon 1.4

Compute Engine

spark 3.3.4

Minimal reproduce step

1. Flink write with op_ts TimestampWLZ type(Microsecond)

Flink SQL> select * from tab_sync_opts_timestamp_1;
+----+------+-----+---------------------+-------------------------+
| id | name | age |         modify_time |                   op_ts |
+----+------+-----+---------------------+-------------------------+
|  7 |    a | 131 | 2026-07-21 06:05:02 | 2026-07-21 20:44:10.000 |
+----+------+-----+---------------------+-------------------------+
1 row in set (3.13 seconds)

2. spark read with Timestamp (Milliseconds)

spark-sql> select * from tab_sync_opts_timestamp_1;
7        a        131        2026-07-21 06:05:02        2026-07-21 12:44:10
Time taken: 0.041 seconds, Fetched 1 row(s)

// Equivalent query without output
spark-sql> select * from tab_sync_opts_timestamp_1 where op_ts = '2026-07-21 12:44:10' ;
Time taken: 0.027 seconds

What doesn't meet your expectations?

Timestamp predicates are converted using the current Paimon field precision
without validating the logical timestamp type in the actual Parquet file.

Both TIMESTAMP_MILLIS and TIMESTAMP_MICROS use the Parquet INT64 physical type,
but their values use different units. Pushing a millisecond predicate into a
microsecond file can therefore filter out matching rows and produce incorrect
query results.

Parquet predicate pushdown should be skipped when the file timestamp unit or
the adjustedToUTC flag does not match the current Paimon field type.

After validating the Parquet timestamp logical type and disabling incompatible
pushdown, the results are correct:

= returns the row
= returns the row
returns no rows
< returns no rows

Anything else?

No

Are you willing to submit a PR?

• I'm willing to submit a PR!