[SUPPORT] The requested schema is not compatible with the file schema

**ClusteringOperator read parquet file erro with incompatible types**

**FLINK SQL**
```
DROP TABLE IF EXISTS stream_catalog.mts_moirai.bl_mts_moirai_audit_center_inspection_hudi_detail;
CREATE TABLE stream_catalog.mts_moirai.bl_mts_moirai_audit_center_inspection_hudi_detail(
App STRING,
AppId STRING,
business_name STRING,
business_key STRING,
ticket_id STRING,
queue_name STRING,
queue_key STRING,
report_id STRING,
remote_id STRING,
in_time STRING,
audit_time STRING,
monitor_time STRING,
listen_time STRING,
last_auditor_account STRING,
last_audit_time STRING,
auditor_name STRING,
auditor_group STRING,
auditor_id STRING,
audit_action_name STRING,
audit_action_key STRING,
is_spam_feedback INT,
is_security_feedback INT,
is_fraust_feedback INT,
is_review INT,
review_queue STRING,
review_audit_count INT,
review_recheck_time INT,
is_inspection INT,
is_appeal INT,
final_action STRING,
is_test INT,
audit_content STRING,
review_content STRING,
inspection_content STRING,
appeal_content STRING,
event_timestamp TIMESTAMP,
partition_date STRING,
is_inspection_appeal STRING,
is_mark STRING,
PRIMARY KEY (event_timestamp) NOT ENFORCED
),
PARTITIONED BY (partition_date) WITH (
'connector' = 'hudi',
'path' = 'hdfs://router/user/mts_moirai/bl_mts_moirai_audit_center_inspection_hudi_detail',
'table.type' = 'COPY_ON_WRITE',
'hoodie.payload.event.time.field' = 'event_timestamp',
'write.operation' = 'insert',
'write.tasks' = '3',
'clustering.tasks' = '3',
'clustering.schedule.enabled' = 'true',
'clustering.async.enabled' = 'true',
'clustering.delta_commits' = '20',
'hive_sync.enable' = 'true',
'hive_sync.mode' = 'hiveql',
'hive_sync.db' = 'mts_moirai',
'hive_sync.table' = 'bl_mts_moirai_audit_center_inspection_hudi_detail'
);

INSERT INTO stream_catalog.mts_moirai.bl_mts_moirai_audit_center_inspection_hudi_detail
SELECT
App
,AppId
,business_name
,business_key
,ticket_id
,queue_name
,queue_key
,report_id
,remote_id
,in_time
,audit_time
,monitor_time
,listen_time
,last_auditor_account
,last_audit_time
,auditor_name
,auditor_group
,auditor_group
,auditor_id
,audit_action_name
,audit_action_key
,is_spam_feedback
,is_security_feedback
,is_fraust_feedback
,is_review
,review_queue
,review_audit_count
,review_recheck_time
,is_inspection
,is_appeal
,final_action
,is_test
,audit_content
,review_content
,inspection_content
,appeal_content
,TO_TIMESTAMP(event_time, 'yyyy-MM-dd HH:mm:ss') AS event_timestamp
,date_format(event_time, 'yyyyMMdd') AS partition_date
,is_inspection_appeal AS is_inspection_appeal
,is_mark AS is_mark
FROM (
SELECT
(CASE WHEN TRIM(App) IS NOT NULL THEN TRIM(App) ELSE '' END) AS App
,(CASE WHEN TRIM(AppId) IS NOT NULL THEN TRIM(AppId) ELSE '' END) AS AppId
,(CASE WHEN TRIM(business_name) IS NOT NULL THEN TRIM(business_name) ELSE '' END) AS business_name
,(CASE WHEN TRIM(business_key) IS NOT NULL THEN TRIM(business_key) ELSE '' END) AS business_key
,(CASE WHEN TRIM(ticket_id) IS NOT NULL THEN TRIM(ticket_id) ELSE '' END) AS ticket_id
,(CASE WHEN TRIM(queue_name) IS NOT NULL THEN TRIM(queue_name) ELSE '' END) AS queue_name
,(CASE WHEN TRIM(queue_key) IS NOT NULL THEN queue_key ELSE '' END) AS queue_key
,(CASE WHEN TRIM(report_id) IS NOT NULL THEN report_id ELSE '' END) AS report_id
,(CASE WHEN TRIM(remote_id) IS NOT NULL THEN remote_id ELSE '' END) AS remote_id
,(CASE WHEN TRIM(in_time) IS NOT NULL THEN in_time ELSE '' END) AS in_time
,(CASE WHEN TRIM(audit_time) IS NOT NULL THEN audit_time ELSE '' END) AS audit_time
,(CASE WHEN TRIM(monitor_time) IS NOT NULL THEN monitor_time ELSE '' END) AS monitor_time
,(CASE WHEN TRIM(listen_time) IS NOT NULL THEN listen_time ELSE '' END) AS listen_time
,(CASE WHEN TRIM(last_auditor_account) IS NOT NULL THEN last_auditor_account ELSE '' END) AS last_auditor_account
,(CASE WHEN TRIM(last_audit_time) IS NOT NULL THEN last_audit_time ELSE '' END) AS last_audit_time
,(CASE WHEN TRIM(auditor_name) IS NOT NULL THEN auditor_name ELSE '' END) AS auditor_name
,(CASE WHEN TRIM(auditor_group) IS NOT NULL THEN auditor_group ELSE '' END) AS auditor_group
,(CASE WHEN TRIM(auditor_id) IS NOT NULL THEN auditor_id ELSE '' END) AS auditor_id
,(CASE WHEN TRIM(audit_action_name) IS NOT NULL THEN audit_action_name ELSE '' END) AS audit_action_name
,(CASE WHEN TRIM(audit_action_key) IS NOT NULL THEN audit_action_key ELSE '' END) AS audit_action_key
,(CASE WHEN is_spam_feedback IS NOT NULL THEN is_spam_feedback ELSE 0 END) AS is_spam_feedback
,(CASE WHEN is_security_feedback IS NOT NULL THEN is_security_feedback ELSE 0 END) AS is_security_feedback
,(CASE WHEN is_fraust_feedback IS NOT NULL THEN is_fraust_feedback ELSE 0 END) AS is_fraust_feedback
,(CASE WHEN is_review IS NOT NULL THEN is_review ELSE 0 END) AS is_review
,(CASE WHEN TRIM(review_queue) IS NOT NULL THEN review_queue ELSE '' END) AS review_queue
,(CASE WHEN review_audit_count IS NOT NULL THEN review_audit_count ELSE 0 END) AS review_audit_count
,(CASE WHEN review_recheck_time IS NOT NULL THEN review_recheck_time ELSE 0 END) AS review_recheck_time
,(CASE WHEN is_inspection IS NOT NULL THEN is_inspection ELSE 0 END) AS is_inspection
,(CASE WHEN is_appeal IS NOT NULL THEN is_appeal ELSE 0 END) AS is_appeal
,(CASE WHEN TRIM(final_action) IS NOT NULL THEN final_action ELSE '' END) AS final_action
,(CASE WHEN is_test IS NOT NULL THEN is_test ELSE 0 END) AS is_test
,(CASE WHEN TRIM(audit_content) IS NOT NULL THEN audit_content ELSE '' END) AS audit_content
,(CASE WHEN TRIM(review_content) IS NOT NULL THEN review_content ELSE '' END) AS review_content
,(CASE WHEN TRIM(inspection_content) IS NOT NULL THEN inspection_content ELSE '' END) AS inspection_content
,(CASE WHEN TRIM(appeal_content) IS NOT NULL THEN appeal_content ELSE '' END) AS appeal_content
,(CASE WHEN TRIM(event_time) IS NOT NULL THEN event_time ELSE '2011-01-01 00:00:00' END) AS event_time
,(CASE WHEN is_inspection_appeal IS NOT NULL THEN is_inspection_appeal ELSE '0' END) AS is_inspection_appeal
,(CASE WHEN is_mark IS NOT NULL THEN is_mark ELSE '0' END) AS is_mark
FROM stream_catalog.mts_moirai.stl_audit_center_inspection_kafka_log
) inspection_log
;
```

**ERROR LOG**
'''
2025-04-24 12:34:58,857 ERROR org.apache.hudi.sink.clustering.ClusteringOperator [] - Executor executes action [Execute clustering for instant 20241226135141790 from task 2] error
org.apache.hudi.exception.HoodieException: unable to read next record from parquet file
at org.apache.hudi.common.util.ParquetReaderIterator.hasNext(ParquetReaderIterator.java:53) ~[hudi-flink1.14-bundle-0.13.0.1.jar:0.13.0.1]
at org.apache.hudi.common.util.MappingIterator.hasNext(MappingIterator.java:35) ~[hudi-flink1.14-bundle-0.13.0.1.jar:0.13.0.1]
at org.apache.hudi.common.util.MappingIterator.hasNext(MappingIterator.java:35) ~[hudi-flink1.14-bundle-0.13.0.1.jar:0.13.0.1]
at java.util.Spliterators$IteratorSpliterator.tryAdvance(Spliterators.java:1811) ~[?:1.8.0_121]
at java.util.stream.StreamSpliterators$WrappingSpliterator.lambda$initPartialTraversalState$0(StreamSpliterators.java:294) ~[?:1.8.0_121]
at java.util.stream.StreamSpliterators$AbstractWrappingSpliterator.fillBuffer(StreamSpliterators.java:206) ~[?:1.8.0_121]
at java.util.stream.StreamSpliterators$AbstractWrappingSpliterator.doAdvance(StreamSpliterators.java:161) ~[?:1.8.0_121]
at java.util.stream.StreamSpliterators$WrappingSpliterator.tryAdvance(StreamSpliterators.java:300) ~[?:1.8.0_121]
at java.util.Spliterators$1Adapter.hasNext(Spliterators.java:681) ~[?:1.8.0_121]
at org.apache.hudi.client.utils.ConcatenatingIterator.hasNext(ConcatenatingIterator.java:45) ~[hudi-flink1.14-bundle-0.13.0.1.jar:0.13.0.1]
at org.apache.hudi.sink.clustering.ClusteringOperator.doClustering(ClusteringOperator.java:261) ~[hudi-flink1.14-bundle-0.13.0.1.jar:0.13.0.1]
at org.apache.hudi.sink.clustering.ClusteringOperator.lambda$processElement$0(ClusteringOperator.java:194) ~[hudi-flink1.14-bundle-0.13.0.1.jar:0.13.0.1]
at org.apache.hudi.sink.utils.NonThrownExecutor.lambda$wrapAction$0(NonThrownExecutor.java:130) ~[hudi-flink1.14-bundle-0.13.0.1.jar:0.13.0.1]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) [?:1.8.0_121]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) [?:1.8.0_121]
at java.lang.Thread.run(Thread.java:745) [?:1.8.0_121]
Caused by: org.apache.parquet.io.ParquetDecodingException: Can not read value at 0 in block -1 in file hdfs://router/user/mts_moirai/bl_mts_moirai_audit_center_inspection_hudi_detail/20241226/d3badaf6-c17e-4389-bf71-261222a59f68-0_1-3-0_20241226133152412.parquet
at org.apache.parquet.hadoop.InternalParquetRecordReader.nextKeyValue(InternalParquetRecordReader.java:255) ~[flink-sql-parquet_2.11-1.14.3.jar:1.14.3]
at org.apache.parquet.hadoop.ParquetReader.read(ParquetReader.java:132) ~[flink-sql-parquet_2.11-1.14.3.jar:1.14.3]
at org.apache.parquet.hadoop.ParquetReader.read(ParquetReader.java:136) ~[flink-sql-parquet_2.11-1.14.3.jar:1.14.3]
at org.apache.hudi.common.util.ParquetReaderIterator.hasNext(ParquetReaderIterator.java:48) ~[hudi-flink1.14-bundle-0.13.0.1.jar:0.13.0.1]
... 15 more
Caused by: org.apache.parquet.io.ParquetDecodingException: The requested schema is not compatible with the file schema. incompatible types: required int64 event_timestamp (TIMESTAMP(MICROS,true)) != required int96 event_timestamp
at org.apache.parquet.io.ColumnIOFactory$ColumnIOCreatorVisitor.incompatibleSchema(ColumnIOFactory.java:101) ~[flink-sql-parquet_2.11-1.14.3.jar:1.14.3]
at org.apache.parquet.io.ColumnIOFactory$ColumnIOCreatorVisitor.visit(ColumnIOFactory.java:93) ~[flink-sql-parquet_2.11-1.14.3.jar:1.14.3]
at org.apache.parquet.schema.PrimitiveType.accept(PrimitiveType.java:596) ~[flink-sql-parquet_2.11-1.14.3.jar:1.14.3]
at org.apache.parquet.io.ColumnIOFactory$ColumnIOCreatorVisitor.visitChildren(ColumnIOFactory.java:83) ~[flink-sql-parquet_2.11-1.14.3.jar:1.14.3]
at org.apache.parquet.io.ColumnIOFactory$ColumnIOCreatorVisitor.visit(ColumnIOFactory.java:57) ~[flink-sql-parquet_2.11-1.14.3.jar:1.14.3]
at org.apache.parquet.schema.MessageType.accept(MessageType.java:55) ~[flink-sql-parquet_2.11-1.14.3.jar:1.14.3]
at org.apache.parquet.io.ColumnIOFactory.getColumnIO(ColumnIOFactory.java:162) ~[flink-sql-parquet_2.11-1.14.3.jar:1.14.3]
at org.apache.parquet.hadoop.InternalParquetRecordReader.checkRead(InternalParquetRecordReader.java:136) ~[flink-sql-parquet_2.11-1.14.3.jar:1.14.3]
at org.apache.parquet.hadoop.InternalParquetRecordReader.nextKeyValue(InternalParquetRecordReader.java:226) ~[flink-sql-parquet_2.11-1.14.3.jar:1.14.3]
at org.apache.parquet.hadoop.ParquetReader.read(ParquetReader.java:132) ~[flink-sql-parquet_2.11-1.14.3.jar:1.14.3]
at org.apache.parquet.hadoop.ParquetReader.read(ParquetReader.java:136) ~[flink-sql-parquet_2.11-1.14.3.jar:1.14.3]
at org.apache.hudi.common.util.ParquetReaderIterator.hasNext(ParquetReaderIterator.java:48) ~[hudi-flink1.14-bundle-0.13.0.1.jar:0.13.0.1]
... 15 more
'''

flink version : 1.14.3
hudi version: 0.13.0.1

Due to this error, the merge operation could not be executed and a very large number of files have been generated. Is there a solution?

Start with org.apache.hudi.sink.clustering.ClusteringOperator.java and the ParquetReaderIterator stack frames, then reproduce clustering with Flink 1.14.3 and Hudi 0.13.0.1 using the reported event_timestamp schemas. Done means the affected parquet file can be read during clustering and the merge completes without the schema-compatibility exception.

Written by the indexing model from the issue text.