fix: reject Parquet timestamp columns read as non-timestamp types

native/core/src/parquet/eager_page_index_reader_factory.rs

@@ -96,7 +96,8 @@
 use crate::parquet::datetime_rebase::stamp_int96_leaves;
 use crate::parquet::parquet_support::SparkParquetOptions;
 use crate::parquet::schema_adapter::{
-    check_file_conversions, RejectOnNonEmpty, NANOS_TIMESTAMP_KEY, REPEATED_PRIMITIVE_KEY,
+    check_file_conversions, RejectOnNonEmpty, INT96_TIMESTAMP_KEY, NANOS_TIMESTAMP_KEY,
+    REPEATED_PRIMITIVE_KEY,
 };
 use arrow::datatypes::{DataType, FieldRef, Schema, SchemaRef};
 use async_trait::async_trait;
@@ -540,10 +541,15 @@ fn is_nanos_timestamp(column: &ColumnDescPtr) -> bool {
     )
 }
 
+fn is_int96(column: &ColumnDescPtr) -> bool {
+    column.physical_type() == parquet::basic::Type::INT96
+}
+
 /// Arrow normalizes both standard and legacy LISTs to the same type, and gives INT96 and
-/// `TIMESTAMP(NANOS)` leaves the same nanosecond timestamp type. Carry the repeated primitive
-/// and `TIMESTAMP(NANOS)` distinctions in an ephemeral schema used by the conversion check,
-/// never in the shared footer cache or in the schema given to the decoder.
+/// `TIMESTAMP(NANOS)` leaves the same nanosecond timestamp type (and INT96 coercion later makes
+/// INT96 look like an INT64 timestamp). Carry the repeated primitive, `TIMESTAMP(NANOS)` and
+/// INT96 distinctions in an ephemeral schema used by the conversion check, never in the shared
+/// footer cache or in the schema given to the decoder.
 fn mark_parquet_leaves(schema: &Schema, parquet: &SchemaDescriptor) -> Schema {
     fn mark(field: &FieldRef, columns: &[ColumnDescPtr], index: &mut usize) -> FieldRef {
         let data_type = match field.data_type() {
@@ -570,15 +576,19 @@ fn mark_parquet_leaves(schema: &Schema, parquet: &SchemaDescriptor) -> Schema {
                         == parquet::basic::Repetition::REPEATED
                 });
                 let nanos = column.is_some_and(is_nanos_timestamp);
+                let int96 = column.is_some_and(is_int96);
                 let mut field = field.as_ref().clone();
-                if repeated || nanos {
+                if repeated || nanos || int96 {
                     let mut metadata = field.metadata().clone();
                     if repeated {
                         metadata.insert(REPEATED_PRIMITIVE_KEY.to_string(), "true".to_string());
                     }
                     if nanos {
                         metadata.insert(NANOS_TIMESTAMP_KEY.to_string(), "true".to_string());
                     }
+                    if int96 {
+                        metadata.insert(INT96_TIMESTAMP_KEY.to_string(), "true".to_string());
+                    }
                     field = field.with_metadata(metadata);
                 }
                 return Arc::new(field);
@@ -774,11 +784,14 @@ impl AsyncFileReader for EagerPageIndexReader {
             {
                 let file = metadata.file_metadata();
                 let columns = file.schema_descr().columns();
-                // Only this check sees the Parquet type of a nanosecond timestamp leaf, so a
-                // `TIMESTAMP(NANOS)` rejection is deferred from here even without row filters.
-                let has_nanos = columns.iter().any(is_nanos_timestamp);
+                // Only this check sees the Parquet type of a timestamp leaf, so the
+                // `TIMESTAMP(NANOS)` and INT96 rejections are deferred from here even without
+                // row filters.
+                let has_parquet_typed_leaf = columns
+                    .iter()
+                    .any(|column| is_nanos_timestamp(column) || is_int96(column));
                 let needs_check = *row_filters
-                    || has_nanos
+                    || has_parquet_typed_leaf
                     || columns.iter().any(|column| {
                         column.self_type().get_basic_info().repetition()
                             == parquet::basic::Repetition::REPEATED
@@ -800,7 +813,7 @@ impl AsyncFileReader for EagerPageIndexReader {
                     DataFusionError::External(source) => ParquetError::External(source),
                     other => ParquetError::External(Box::new(other)),
                 })?;
-                if *row_filters || has_nanos {
+                if *row_filters || has_parquet_typed_leaf {
                     if let Some(rejection) = &rejection {
                         self.deferred_rejections.lock().unwrap().insert(
                             object_meta.location.clone(),