fix(parquet): add reader option for unknown physical/logical type combos

parquet/src/arrow/arrow_reader/mod.rs

@@ -826,6 +826,18 @@ impl ArrowReaderOptions {
         self
     }
 
+    /// Treat incompatible physical/logical type combinations as an unknown
+    /// logical type when reading (parquet-format GH-607).
+    ///
+    /// Default is `false`: such combinations return an error. When `true`, the
+    /// logical type is rewritten to `_Unknown` with sort order `UNDEFINED`.
+    /// Column statistics are retained.
+    pub fn with_coerce_incompatible_logical_types(mut self, coerce: bool) -> Self {
+        self.metadata_options
+            .set_coerce_incompatible_logical_types(coerce);
+        self
+    }
+
     /// Provide the file decryption properties to use when reading encrypted parquet files.
     ///
     /// If encryption is enabled and the file is encrypted, the `file_decryption_properties` must be provided.

parquet/src/arrow/schema/primitive.rs

@@ -183,6 +183,7 @@ fn check_decimal_length(type_length: i32) -> Result<()> {
 fn from_int32(info: &BasicTypeInfo, scale: i32, precision: i32) -> Result<DataType> {
     match (info.logical_type_ref(), info.converted_type()) {
         (None, ConvertedType::NONE) => Ok(DataType::Int32),
+        (Some(LogicalType::_Unknown { .. }), _) => Ok(DataType::Int32),
         (Some(t @ LogicalType::Integer(int)), _) => match (int.bit_width, int.is_signed) {
             (8, true) => Ok(DataType::Int8),
             (16, true) => Ok(DataType::Int16),
@@ -223,6 +224,7 @@ fn from_int32(info: &BasicTypeInfo, scale: i32, precision: i32) -> Result<DataTy
 fn from_int64(info: &BasicTypeInfo, scale: i32, precision: i32) -> Result<DataType> {
     match (info.logical_type_ref(), info.converted_type()) {
         (None, ConvertedType::NONE) => Ok(DataType::Int64),
+        (Some(LogicalType::_Unknown { .. }), _) => Ok(DataType::Int64),
         (
             Some(LogicalType::Integer(IntType {
                 bit_width: 64,
@@ -457,4 +459,37 @@ mod tests {
             DataType::Interval(IntervalUnit::DayTime)
         );
     }
+
+    fn coerced_unknown(physical: PhysicalType) -> Type {
+        Type::primitive_type_builder("c", physical)
+            .with_repetition(Repetition::REQUIRED)
+            .with_logical_type(Some(LogicalType::Uuid))
+            .with_coerce_incompatible_logical_types(true)
+            .build()
+            .unwrap()
+    }
+
+    #[test]
+    fn unknown_logical_type_on_int32_is_int32() {
+        assert_eq!(
+            convert_primitive(&coerced_unknown(PhysicalType::INT32), None).unwrap(),
+            DataType::Int32
+        );
+    }
+
+    #[test]
+    fn unknown_logical_type_on_int64_is_int64() {
+        assert_eq!(
+            convert_primitive(&coerced_unknown(PhysicalType::INT64), None).unwrap(),
+            DataType::Int64
+        );
+    }
+
+    #[test]
+    fn unknown_logical_type_on_byte_array_is_binary() {
+        assert_eq!(
+            convert_primitive(&coerced_unknown(PhysicalType::BYTE_ARRAY), None).unwrap(),
+            DataType::Binary
+        );
+    }
 }

parquet/src/file/metadata/options.rs

@@ -94,6 +94,7 @@ pub struct ParquetMetaDataOptions {
     encoding_stats_policy: ParquetStatisticsPolicy,
     column_stats_policy: ParquetStatisticsPolicy,
     size_stats_policy: ParquetStatisticsPolicy,
+    coerce_incompatible_logical_types: bool,
 }
 
 impl Default for ParquetMetaDataOptions {
@@ -104,6 +105,7 @@ impl Default for ParquetMetaDataOptions {
             encoding_stats_policy: ParquetStatisticsPolicy::KeepAll,
             column_stats_policy: ParquetStatisticsPolicy::KeepAll,
             size_stats_policy: ParquetStatisticsPolicy::KeepAll,
+            coerce_incompatible_logical_types: false,
         }
     }
 }
@@ -244,6 +246,29 @@ impl ParquetMetaDataOptions {
         self.set_size_stats_policy(policy);
         self
     }
+
+    /// Returns whether incompatible physical/logical type combinations should
+    /// be treated as an unknown logical type when reading.
+    ///
+    /// Default is `false`: such combinations return an error, matching historical
+    /// behavior. When `true`, the logical type is rewritten to `_Unknown` with
+    /// sort order `UNDEFINED`. Column statistics are retained.
+    pub fn coerce_incompatible_logical_types(&self) -> bool {
+        self.coerce_incompatible_logical_types
+    }
+
+    /// Sets whether to coerce incompatible physical/logical type combinations.
+    ///
+    /// See [`Self::coerce_incompatible_logical_types`].
+    pub fn set_coerce_incompatible_logical_types(&mut self, val: bool) {
+        self.coerce_incompatible_logical_types = val;
+    }
+
+    /// Call [`Self::set_coerce_incompatible_logical_types`] and return `Self` for chaining.
+    pub fn with_coerce_incompatible_logical_types(mut self, val: bool) -> Self {
+        self.set_coerce_incompatible_logical_types(val);
+        self
+    }
 }
 
 #[cfg(test)]

parquet/src/file/metadata/thrift/mod.rs

@@ -55,6 +55,7 @@ use crate::{
     },
     schema::types::{
         ColumnDescriptor, SchemaDescriptor, TypePtr, num_nodes, parquet_schema_from_array,
+        parquet_schema_from_array_opts,
     },
     thrift_struct,
     util::bit_util::FromBytes,
@@ -805,7 +806,10 @@ pub(crate) fn parquet_metadata_from_bytes(
                     // read schema and convert to SchemaDescriptor for use when reading row groups
                     let val =
                         read_thrift_vec::<SchemaElement, ThriftSliceInputProtocol>(&mut prot)?;
-                    let val = parquet_schema_from_array(val)?;
+                    let coerce = options
+                        .map(|o| o.coerce_incompatible_logical_types())
+                        .unwrap_or(false);
+                    let val = parquet_schema_from_array_opts(val, coerce)?;
                     schema_descr = Some(Arc::new(SchemaDescriptor::new(val)));
                 }
             }

parquet/src/file/serialized_reader.rs

@@ -204,6 +204,18 @@ impl ReadOptionsBuilder {
         self
     }
 
+    /// Treat incompatible physical/logical type combinations as an unknown
+    /// logical type when reading (parquet-format GH-607).
+    ///
+    /// Default is `false`: such combinations return an error. When `true`, the
+    /// logical type is rewritten to `_Unknown` with sort order `UNDEFINED`.
+    /// Column statistics are retained.
+    pub fn with_coerce_incompatible_logical_types(mut self, coerce: bool) -> Self {
+        self.metadata_options
+            .set_coerce_incompatible_logical_types(coerce);
+        self
+    }
+
     /// Seal the builder and return the read options
     pub fn build(self) -> ReadOptions {
         let props = self
@@ -2963,4 +2975,60 @@ mod tests {
         }
         assert_eq!(num_rows, reader.metadata().file_metadata().num_rows());
     }
+
+    #[test]
+    fn test_int32_uuid_logical_type_errors_by_default() {
+        let file = get_test_file("int32_with_uuid_logical_type.parquet");
+        let err = SerializedFileReader::new(file)
+            .err()
+            .expect("default should reject INT32+UUID");
+        assert!(
+            err.to_string()
+                .contains("Cannot annotate Uuid from INT32 for field 'int32_uuid'"),
+            "{err}"
+        );
+    }
+
+    #[test]
+    fn test_int32_uuid_logical_type_coerced_with_option() {
+        let file = get_test_file("int32_with_uuid_logical_type.parquet");
+        let options = ReadOptionsBuilder::new()
+            .with_coerce_incompatible_logical_types(true)
+            .build();
+        let reader = SerializedFileReader::new_with_options(file, options).unwrap();
+
+        let schema = reader.metadata().file_metadata().schema_descr();
+        assert_eq!(schema.column(0).name(), "int32_uuid");
+        assert_eq!(schema.column(0).physical_type(), Type::INT32);
+        assert_eq!(
+            schema.column(0).logical_type_ref(),
+            Some(&basic::LogicalType::_Unknown { field_id: 0 })
+        );
+        assert_eq!(schema.column(0).sort_order(), SortOrder::UNDEFINED);
+        assert_eq!(
+            reader
+                .metadata()
+                .file_metadata()
+                .column_order(0)
+                .sort_order(),
+            SortOrder::UNDEFINED
+        );
+        assert!(
+            reader
+                .metadata()
+                .row_group(0)
+                .column(0)
+                .statistics()
+                .is_some()
+        );
+
+        let mut iter = reader
+            .get_row_iter(None)
+            .expect("Failed to create row iterator");
+        let mut num_rows = 0;
+        while iter.next().is_some() {
+            num_rows += 1;
+        }
+        assert_eq!(num_rows, 10);
+    }
 }