[BUG] Spark will read invalid timestamp(3) data when record in log is older than the same in parquet.

Describe the problem you faced

Spark read invalid timestamp(3) data when record in log is older than the same in parquet.

To Reproduce

1. create a mor table with timestamp(3) type.
eg.
CREATE EXTERNAL TABLE `xxx.bucket_mor_t2`(
`_hoodie_commit_time` string COMMENT '',
`_hoodie_commit_seqno` string COMMENT '',
`_hoodie_record_key` string COMMENT '',
`_hoodie_partition_path` string COMMENT '',
`_hoodie_file_name` string COMMENT '',
`source_from` int COMMENT '',
`id` bigint COMMENT '',
`name` string COMMENT '',
`create_time` timestamp COMMENT '',
`price` decimal(14,2) COMMENT '',
`extend` string COMMENT '',
`count` bigint COMMENT '',
`create_date` date COMMENT '',
`ext_dt` timestamp COMMENT '',
`precombine_field` string COMMENT '',
`sync_deleted` int COMMENT '',
`sync_time` timestamp COMMENT '',
`__binlog_file` string COMMENT '',
`__pos` int COMMENT '',
`source_sys` int COMMENT '')
PARTITIONED BY (
`__partition_field` int COMMENT '')
ROW FORMAT SERDE
'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
WITH SERDEPROPERTIES (
'hoodie.query.as.ro.table'='false',
'path'='hdfs://NameNodeService1/xxx/xxx/bucket_mor_t2')
STORED AS INPUTFORMAT
'org.apache.hudi.hadoop.realtime.HoodieParquetRealtimeInputFormat'
OUTPUTFORMAT
'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'
LOCATION
'hdfs://NameNodeService1/xxx/xxx/bucket_mor_t2'
TBLPROPERTIES (
'connector'='hudi',
'hoodie.datasource.write.recordkey.field'='source_from,id',
'last_commit_time_sync'='20231106172508127',
'path'='hdfs://NameNodeService1/xxx/xxx/bucket_mor_t2',
'spark.sql.sources.provider'='hudi',
'spark.sql.sources.schema.numPartCols'='1',
'spark.sql.sources.schema.numParts'='1',
'spark.sql.sources.schema.part.0'='{"type":"struct","fields":[{"name":"_hoodie_commit_time","type":"string","nullable":true,"metadata":{}},{"name":"_hoodie_commit_seqno","type":"string","nullable":true,"metadata":{}},{"name":"_hoodie_record_key","type":"string","nullable":true,"metadata":{}},{"name":"_hoodie_partition_path","type":"string","nullable":true,"metadata":{}},{"name":"_hoodie_file_name","type":"string","nullable":true,"metadata":{}},{"name":"source_from","type":"integer","nullable":true,"metadata":{}},{"name":"id","type":"long","nullable":true,"metadata":{}},{"name":"name","type":"string","nullable":true,"metadata":{}},{"name":"create_time","type":"timestamp","nullable":true,"metadata":{}},{"name":"price","type":"decimal(14,2)","nullable":true,"metadata":{}},{"name":"extend","type":"string","nullable":true,"metadata":{}},{"name":"count","type":"long","nullable":true,"metadata":{}},{"name":"create_date","type":"date","nullable":true,"metadata":{}},{"name":"ext_dt","type":"timestamp","nullable":true,"metadata":{}},{"name":"precombine_field","type":"string","nullable":true,"metadata":{}},{"name":"sync_deleted","type":"integer","nullable":true,"metadata":{}},{"name":"sync_time","type":"timestamp","nullable":true,"metadata":{}},{"name":"__binlog_file","type":"string","nullable":true,"metadata":{}},{"name":"__pos","type":"integer","nullable":true,"metadata":{}},{"name":"source_sys","type":"integer","nullable":true,"metadata":{}},{"name":"__partition_field","type":"integer","nullable":true,"metadata":{}}]}',
'spark.sql.sources.schema.partCol.0'='__partition_field',
'table.type'='MERGE_ON_READ',
'transient_lastDdlTime'='1692251328')

2. insert new data into parquet with flink engine. eg. insert a record(id=1) with precombine value = 000001308800000028038927500000

3. mock binlog(same record in step2) with precombine value = 1 (which is smaller than before) and commit but don't do compaction

finally, read record(id=1) in snapthot mode with spark sql. invalid data will occur:


Expected behavior

when read field with timestamp(3), it will occur invalid data.

Environment Description

- Hudi version : 0.12.1
- Spark version : 3.1.1

Additional context
here is some debug info:
create_time in parquet is micros

avro log is millis

Stacktrace

The issue names no source files or tests. Start by reproducing the MERGE_ON_READ snapshot read with Spark 3.1.1, comparing the parquet timestamp in micros with the Avro log timestamp in millis; done means the older log record no longer produces invalid timestamp data.

Written by the indexing model from the issue text.