From c7be8c92ab015d15fd3782ae26fd95a5ba51f1d6 Mon Sep 17 00:00:00 2001 From: Marcelo Tesla <9055877+M-Tesla@users.noreply.github.com> Date: Mon, 7 Sep 2026 13:11:39 -0300 Subject: [PATCH 1/2] fix(parquet): add reader option for unknown physical/logical type combos parquet-format GH-607 requires readers to treat unrecognized physical and logical type combinations as an unknown logical type. Keep the historical error as the default, and add a reader option that exposes the physical type and ignores column statistics. --- parquet/src/arrow/arrow_reader/mod.rs | 99 ++++++- parquet/src/file/metadata/options.rs | 25 ++ parquet/src/file/metadata/thrift/mod.rs | 13 +- parquet/src/file/serialized_reader.rs | 12 + parquet/src/schema/types.rs | 330 ++++++++++++++++++++---- 5 files changed, 430 insertions(+), 49 deletions(-) diff --git a/parquet/src/arrow/arrow_reader/mod.rs b/parquet/src/arrow/arrow_reader/mod.rs index 6cc07ffbf52a..5635e1812682 100644 --- a/parquet/src/arrow/arrow_reader/mod.rs +++ b/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 + /// column is exposed as its physical type with no logical annotation, and + /// column statistics are ignored. + 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. @@ -1809,7 +1821,7 @@ pub(crate) mod tests { virtual_type::{RowGroupIndex, RowNumber}, }; use crate::arrow::{ArrowWriter, ProjectionMask}; - use crate::basic::{ConvertedType, Encoding, Repetition, Type as PhysicalType}; + use crate::basic::{ConvertedType, Encoding, LogicalType, Repetition, Type as PhysicalType}; use crate::column::reader::decoder::REPETITION_LEVELS_BATCH_SIZE; use crate::data_type::{ BoolType, ByteArray, ByteArrayType, DataType, DoubleType, FixedLenByteArray, @@ -6155,4 +6167,89 @@ pub(crate) mod tests { (Bytes::from(buf), metadata) } + + fn parquet_file_with_int32_uuid_logical_type() -> Vec { + let field = Type::primitive_type_builder("id", PhysicalType::INT32) + .with_repetition(Repetition::REQUIRED) + .with_logical_type(Some(LogicalType::Uuid)) + .with_skip_logical_physical_validation(true) + .build() + .unwrap(); + let schema = Arc::new( + Type::group_type_builder("schema") + .with_fields(vec![Arc::new(field)]) + .build() + .unwrap(), + ); + let props = Arc::new( + WriterProperties::builder() + .set_statistics_enabled(EnabledStatistics::Chunk) + .build(), + ); + + let mut buf = Vec::new(); + let mut writer = SerializedFileWriter::new(&mut buf, schema, props).unwrap(); + let mut row_group_writer = writer.next_row_group().unwrap(); + let mut col_writer = row_group_writer.next_column().unwrap().unwrap(); + col_writer + .typed::() + .write_batch(&[1, 2, 3], None, None) + .unwrap(); + col_writer.close().unwrap(); + row_group_writer.close().unwrap(); + let written = writer.close().unwrap(); + assert!( + written.row_group(0).column(0).statistics().is_some(), + "test fixture must write column statistics so ignore-on-read can be asserted" + ); + buf + } + + #[test] + fn test_int32_uuid_logical_type_errors_by_default() { + let file = Bytes::from(parquet_file_with_int32_uuid_logical_type()); + let err = ParquetRecordBatchReaderBuilder::try_new(file).unwrap_err(); + assert!( + err.to_string() + .contains("Cannot annotate Uuid from INT32 for field 'id'"), + "{err}" + ); + } + + #[test] + fn test_int32_uuid_logical_type_coerced_with_option() { + let file = Bytes::from(parquet_file_with_int32_uuid_logical_type()); + let options = ArrowReaderOptions::new().with_coerce_incompatible_logical_types(true); + let builder = ParquetRecordBatchReaderBuilder::try_new_with_options(file, options).unwrap(); + + let parquet_col = builder.metadata().file_metadata().schema_descr().column(0); + assert_eq!(parquet_col.physical_type(), PhysicalType::INT32); + assert_eq!(parquet_col.logical_type_ref(), None); + assert!( + parquet_col + .get_basic_info() + .incompatible_logical_type_coerced() + ); + assert!( + builder + .metadata() + .row_group(0) + .column(0) + .statistics() + .is_none(), + "GH-607: statistics for coerced columns must be ignored" + ); + + let schema = builder.schema(); + assert_eq!(schema.field(0).data_type(), &ArrowDataType::Int32); + + let mut reader = builder.build().unwrap(); + let batch = reader.next().unwrap().unwrap(); + assert!(reader.next().is_none()); + assert_eq!(batch.num_rows(), 3); + let values = batch + .column(0) + .as_primitive::(); + assert_eq!(values.values(), &[1, 2, 3]); + } } diff --git a/parquet/src/file/metadata/options.rs b/parquet/src/file/metadata/options.rs index 3f71c5a0f4c3..b677f01f8607 100644 --- a/parquet/src/file/metadata/options.rs +++ b/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 column is exposed as its physical type with no + /// logical annotation, and column statistics are ignored (parquet-format GH-607). + 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)] diff --git a/parquet/src/file/metadata/thrift/mod.rs b/parquet/src/file/metadata/thrift/mod.rs index 8d590079cc03..211789461b9c 100644 --- a/parquet/src/file/metadata/thrift/mod.rs +++ b/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, @@ -437,6 +438,13 @@ fn read_column_metadata( skip_col_stats = opts.skip_column_stats(col_index); skip_size_stats = opts.skip_size_stats(col_index); } + if column + .column_descr + .get_basic_info() + .incompatible_logical_type_coerced() + { + skip_col_stats = true; + } // struct ColumnMetaData { // 1: required Type type @@ -805,7 +813,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::(&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))); } } diff --git a/parquet/src/file/serialized_reader.rs b/parquet/src/file/serialized_reader.rs index bc525a78aba7..b70d12690e92 100644 --- a/parquet/src/file/serialized_reader.rs +++ b/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 + /// column is exposed as its physical type with no logical annotation, and + /// column statistics are ignored. + 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 diff --git a/parquet/src/schema/types.rs b/parquet/src/schema/types.rs index c6927fb71f84..bca369ca6938 100644 --- a/parquet/src/schema/types.rs +++ b/parquet/src/schema/types.rs @@ -259,6 +259,14 @@ pub struct PrimitiveTypeBuilder<'a> { precision: i32, scale: i32, id: Option, + /// When true, incompatible physical/logical combinations are treated as an + /// unknown logical type instead of returning an error. Used when reading + /// files, not when constructing a schema to write. + coerce_incompatible_logical_types: bool, + /// When true, skip the physical/logical compatibility check so tests can + /// emit files that other writers may produce. The logical type is kept. + #[cfg(test)] + skip_logical_physical_validation: bool, } impl<'a> PrimitiveTypeBuilder<'a> { @@ -274,6 +282,9 @@ impl<'a> PrimitiveTypeBuilder<'a> { precision: -1, scale: -1, id: None, + coerce_incompatible_logical_types: false, + #[cfg(test)] + skip_logical_physical_validation: false, } } @@ -325,22 +336,40 @@ impl<'a> PrimitiveTypeBuilder<'a> { Self { id, ..self } } + /// Treat incompatible physical/logical type combinations as an unknown + /// logical type (physical type, no annotation) instead of erroring. + /// + /// This is the parquet-format GH-607 reader behavior. Do not use when + /// building a schema to write; invalid combinations should still fail. + pub(crate) fn with_coerce_incompatible_logical_types(self, value: bool) -> Self { + Self { + coerce_incompatible_logical_types: value, + ..self + } + } + + /// Skip physical/logical compatibility checks, keeping the logical type. + /// + /// Only for tests that need to write a file with an invalid combination. + #[cfg(test)] + pub(crate) fn with_skip_logical_physical_validation(self, value: bool) -> Self { + Self { + skip_logical_physical_validation: value, + ..self + } + } + /// Creates a new `PrimitiveType` instance from the collected attributes. /// Returns `Err` in case of any building conditions are not met. pub fn build(self) -> Result { - let sort_order = ColumnOrder::column_order_for_type( - self.logical_type.as_ref(), - self.converted_type, - self.physical_type, - ) - .sort_order(); let mut basic_info = BasicTypeInfo { name: String::from(self.name), repetition: Some(self.repetition), converted_type: self.converted_type, logical_type: self.logical_type.clone(), id: self.id, - sort_order, + sort_order: SortOrder::SIGNED, + incompatible_logical_type_coerced: false, }; // Check length before logical type, since it is used for logical type validation. @@ -355,21 +384,19 @@ impl<'a> PrimitiveTypeBuilder<'a> { if let Some(logical_type) = &self.logical_type { // If a converted type is populated, check that it is consistent with // its logical type - if self.converted_type != ConvertedType::NONE { - if ConvertedType::from(self.logical_type.clone()) != self.converted_type { - return Err(general_err!( - "Logical type {:?} is incompatible with converted type {} for field '{}'", - logical_type, - self.converted_type, - self.name - )); - } - } else { - // Populate the converted type for backwards compatibility - basic_info.converted_type = self.logical_type.clone().into(); + if self.converted_type != ConvertedType::NONE + && ConvertedType::from(self.logical_type.clone()) != self.converted_type + { + return Err(general_err!( + "Logical type {:?} is incompatible with converted type {} for field '{}'", + logical_type, + self.converted_type, + self.name + )); } + // Check that logical type and physical type are compatible - match (logical_type, self.physical_type) { + let combo_error = match (logical_type, self.physical_type) { (LogicalType::Map | LogicalType::List | LogicalType::File, _) => { return Err(general_err!( "{:?} cannot be applied to a primitive type for field '{}'", @@ -377,7 +404,7 @@ impl<'a> PrimitiveTypeBuilder<'a> { self.name )); } - (LogicalType::Enum, PhysicalType::BYTE_ARRAY) => {} + (LogicalType::Enum, PhysicalType::BYTE_ARRAY) => None, (LogicalType::Decimal(decimal), _) => { // Check that scale and precision are consistent with legacy values if decimal.scale != self.scale { @@ -397,41 +424,48 @@ impl<'a> PrimitiveTypeBuilder<'a> { )); } self.check_decimal_precision_scale()?; + None } - (LogicalType::Date, PhysicalType::INT32) => {} + (LogicalType::Date, PhysicalType::INT32) => None, ( LogicalType::Time(TimeType { unit: TimeUnit::MILLIS, .. }), PhysicalType::INT32, - ) => {} + ) => None, (LogicalType::Time(time), PhysicalType::INT64) => { if time.unit == TimeUnit::MILLIS { - return Err(general_err!( + Some(general_err!( "Cannot use millisecond unit on INT64 type for field '{}'", self.name - )); + )) + } else { + None } } - (LogicalType::Timestamp(_), PhysicalType::INT64) => {} - (LogicalType::Integer(int), PhysicalType::INT32) if int.bit_width <= 32 => {} - (LogicalType::Integer(int), PhysicalType::INT64) if int.bit_width == 64 => {} + (LogicalType::Timestamp(_), PhysicalType::INT64) => None, + (LogicalType::Integer(int), PhysicalType::INT32) if int.bit_width <= 32 => None, + (LogicalType::Integer(int), PhysicalType::INT64) if int.bit_width == 64 => None, // Null type - (LogicalType::Unknown, _) => {} - (LogicalType::String, PhysicalType::BYTE_ARRAY) => {} - (LogicalType::Json, PhysicalType::BYTE_ARRAY) => {} - (LogicalType::Bson, PhysicalType::BYTE_ARRAY) => {} - (LogicalType::Geometry(_), PhysicalType::BYTE_ARRAY) => {} - (LogicalType::Geography(_), PhysicalType::BYTE_ARRAY) => {} - (LogicalType::Uuid, PhysicalType::FIXED_LEN_BYTE_ARRAY) if self.length == 16 => {} + (LogicalType::Unknown, _) => None, + (LogicalType::String, PhysicalType::BYTE_ARRAY) => None, + (LogicalType::Json, PhysicalType::BYTE_ARRAY) => None, + (LogicalType::Bson, PhysicalType::BYTE_ARRAY) => None, + (LogicalType::Geometry(_), PhysicalType::BYTE_ARRAY) => None, + (LogicalType::Geography(_), PhysicalType::BYTE_ARRAY) => None, + (LogicalType::Uuid, PhysicalType::FIXED_LEN_BYTE_ARRAY) if self.length == 16 => { + None + } (LogicalType::Uuid, PhysicalType::FIXED_LEN_BYTE_ARRAY) => { return Err(general_err!( "UUID cannot annotate field '{}' because it is not a FIXED_LEN_BYTE_ARRAY(16) field", self.name )); } - (LogicalType::Float16, PhysicalType::FIXED_LEN_BYTE_ARRAY) if self.length == 2 => {} + (LogicalType::Float16, PhysicalType::FIXED_LEN_BYTE_ARRAY) if self.length == 2 => { + None + } (LogicalType::Float16, PhysicalType::FIXED_LEN_BYTE_ARRAY) => { return Err(general_err!( "FLOAT16 cannot annotate field '{}' because it is not a FIXED_LEN_BYTE_ARRAY(2) field", @@ -439,18 +473,48 @@ impl<'a> PrimitiveTypeBuilder<'a> { )); } // unknown logical type means just use physical type - (LogicalType::_Unknown { .. }, _) => {} - (a, b) => { - return Err(general_err!( - "Cannot annotate {:?} from {} for field '{}'", - a, - b, - self.name - )); + (LogicalType::_Unknown { .. }, _) => None, + (a, b) => Some(general_err!( + "Cannot annotate {:?} from {} for field '{}'", + a, + b, + self.name + )), + }; + + if let Some(err) = combo_error { + if self.coerce_incompatible_logical_types { + // parquet-format GH-607: treat as unknown logical type. + basic_info.logical_type = None; + basic_info.converted_type = self.converted_type; + basic_info.incompatible_logical_type_coerced = true; + } else { + #[cfg(test)] + { + if self.skip_logical_physical_validation { + if self.converted_type == ConvertedType::NONE { + basic_info.converted_type = self.logical_type.clone().into(); + } + } else { + return Err(err); + } + } + #[cfg(not(test))] + return Err(err); } + } else if self.converted_type == ConvertedType::NONE { + // Populate the converted type for backwards compatibility + basic_info.converted_type = self.logical_type.clone().into(); } } + basic_info.sort_order = ColumnOrder::column_order_for_type( + basic_info.logical_type.as_ref(), + basic_info.converted_type, + self.physical_type, + ) + .sort_order(); + match self.converted_type { ConvertedType::NONE => {} ConvertedType::UTF8 | ConvertedType::BSON | ConvertedType::JSON => { @@ -676,6 +740,7 @@ impl<'a> GroupTypeBuilder<'a> { logical_type: self.logical_type.clone(), id: self.id, sort_order: SortOrder::UNDEFINED, + incompatible_logical_type_coerced: false, }; // Populate the converted type if only the logical type is populated if self.logical_type.is_some() && self.converted_type == ConvertedType::NONE { @@ -788,6 +853,9 @@ pub struct BasicTypeInfo { logical_type: Option, id: Option, sort_order: SortOrder, + /// Set when an incompatible logical type was stripped while reading + /// (parquet-format GH-607). Statistics for this column should be ignored. + incompatible_logical_type_coerced: bool, } impl HeapSize for BasicTypeInfo { @@ -849,6 +917,14 @@ impl BasicTypeInfo { pub fn sort_order(&self) -> SortOrder { self.sort_order } + + /// Returns `true` if an incompatible logical type was stripped while reading + /// this field (parquet-format GH-607). + /// + /// Statistics for such columns should be ignored. + pub(crate) fn incompatible_logical_type_coerced(&self) -> bool { + self.incompatible_logical_type_coerced + } } // ---------------------------------------------------------------------- @@ -1424,6 +1500,16 @@ fn check_logical_type(logical_type: Option<&LogicalType>) -> Result<()> { // convert thrift decoded array of `SchemaElement` into this crate's representation of // parquet types. this function consumes `elements`. pub(crate) fn parquet_schema_from_array(elements: Vec>) -> Result { + parquet_schema_from_array_opts(elements, false) +} + +/// Like [`parquet_schema_from_array`], but when `coerce_incompatible_logical_types` +/// is true, unknown physical/logical combinations are treated as an unknown +/// logical type (parquet-format GH-607). +pub(crate) fn parquet_schema_from_array_opts( + elements: Vec>, + coerce_incompatible_logical_types: bool, +) -> Result { let mut index = 0; let num_elements = elements.len(); let mut schema_nodes = Vec::with_capacity(1); // there should only be one element when done @@ -1432,7 +1518,12 @@ pub(crate) fn parquet_schema_from_array(elements: Vec>) -> Res let mut elements = elements.into_iter(); while index < num_elements { - let t = schema_from_array_helper(&mut elements, num_elements, index)?; + let t = schema_from_array_helper( + &mut elements, + num_elements, + index, + coerce_incompatible_logical_types, + )?; index = t.0; schema_nodes.push(t.1); } @@ -1455,6 +1546,7 @@ fn schema_from_array_helper( elements: &mut IntoIter>, num_elements: usize, index: usize, + coerce_incompatible_logical_types: bool, ) -> Result<(usize, TypePtr)> { // Whether or not the current node is root (message type). // There is only one message type node in the schema tree. @@ -1509,7 +1601,8 @@ fn schema_from_array_helper( .with_length(length) .with_precision(precision) .with_scale(scale) - .with_id(field_id); + .with_id(field_id) + .with_coerce_incompatible_logical_types(coerce_incompatible_logical_types); Ok((index + 1, Arc::new(builder.build()?))) } else { let mut builder = Type::group_type_builder(element.name) @@ -1535,7 +1628,12 @@ fn schema_from_array_helper( let mut fields = Vec::with_capacity(usize::try_from(n)?); let mut next_index = index + 1; for _ in 0..n { - let child_result = schema_from_array_helper(elements, num_elements, next_index)?; + let child_result = schema_from_array_helper( + elements, + num_elements, + next_index, + coerce_incompatible_logical_types, + )?; next_index = child_result.0; fields.push(child_result.1); } @@ -2841,4 +2939,142 @@ mod tests { let result = parquet_schema_from_array(elements); assert!(result.unwrap_err().to_string().contains("Integer overflow")); } + + fn int32_uuid_schema_elements<'a>() -> Vec> { + vec![ + SchemaElement { + r#type: None, + type_length: None, + repetition_type: None, + name: "schema", + num_children: Some(1), + converted_type: None, + scale: None, + precision: None, + field_id: None, + logical_type: None, + }, + SchemaElement { + r#type: Some(PhysicalType::INT32), + type_length: None, + repetition_type: Some(Repetition::REQUIRED), + name: "id", + num_children: None, + converted_type: None, + scale: None, + precision: None, + field_id: None, + logical_type: Some(LogicalType::Uuid), + }, + ] + } + + #[test] + fn test_int32_uuid_rejected_by_default() { + let err = parquet_schema_from_array(int32_uuid_schema_elements()).unwrap_err(); + assert!( + err.to_string() + .contains("Cannot annotate Uuid from INT32 for field 'id'"), + "{err}" + ); + } + + #[test] + fn test_int32_uuid_coerced_to_physical_type() { + let schema = parquet_schema_from_array_opts(int32_uuid_schema_elements(), true).unwrap(); + let fields = schema.get_fields(); + assert_eq!(fields.len(), 1); + let col = &fields[0]; + assert_eq!(col.get_physical_type(), PhysicalType::INT32); + assert_eq!(col.get_basic_info().logical_type_ref(), None); + assert_eq!(col.get_basic_info().converted_type(), ConvertedType::NONE); + assert!(col.get_basic_info().incompatible_logical_type_coerced()); + } + + #[test] + fn test_writer_still_rejects_int32_uuid() { + let err = Type::primitive_type_builder("id", PhysicalType::INT32) + .with_repetition(Repetition::REQUIRED) + .with_logical_type(Some(LogicalType::Uuid)) + .build() + .unwrap_err(); + assert!( + err.to_string() + .contains("Cannot annotate Uuid from INT32 for field 'id'"), + "{err}" + ); + } + + #[test] + fn test_coerce_does_not_relax_uuid_wrong_length() { + let elements = vec![ + SchemaElement { + r#type: None, + type_length: None, + repetition_type: None, + name: "schema", + num_children: Some(1), + converted_type: None, + scale: None, + precision: None, + field_id: None, + logical_type: None, + }, + SchemaElement { + r#type: Some(PhysicalType::FIXED_LEN_BYTE_ARRAY), + type_length: Some(15), + repetition_type: Some(Repetition::REQUIRED), + name: "id", + num_children: None, + converted_type: None, + scale: None, + precision: None, + field_id: None, + logical_type: Some(LogicalType::Uuid), + }, + ]; + let err = parquet_schema_from_array_opts(elements, true).unwrap_err(); + assert!( + err.to_string().contains( + "UUID cannot annotate field 'id' because it is not a FIXED_LEN_BYTE_ARRAY(16) field" + ), + "{err}" + ); + } + + #[test] + fn test_coerce_does_not_allow_list_on_primitive() { + let elements = vec![ + SchemaElement { + r#type: None, + type_length: None, + repetition_type: None, + name: "schema", + num_children: Some(1), + converted_type: None, + scale: None, + precision: None, + field_id: None, + logical_type: None, + }, + SchemaElement { + r#type: Some(PhysicalType::INT32), + type_length: None, + repetition_type: Some(Repetition::REQUIRED), + name: "id", + num_children: None, + converted_type: None, + scale: None, + precision: None, + field_id: None, + logical_type: Some(LogicalType::List), + }, + ]; + let err = parquet_schema_from_array_opts(elements, true).unwrap_err(); + assert!( + err.to_string() + .contains("List cannot be applied to a primitive type for field 'id'"), + "{err}" + ); + } } From 237b2246263997b6d125f524d0e34b64fd8d045e Mon Sep 17 00:00:00 2001 From: Marcelo Tesla <9055877+M-Tesla@users.noreply.github.com> Date: Wed, 9 Sep 2026 19:00:35 -0300 Subject: [PATCH 2/2] fix(parquet): coerce unknown type combos to _Unknown with UNDEFINED sort Keep column statistics, switch tests to the parquet-testing INT32+UUID file, and compute sort order from _Unknown instead of a SIGNED placeholder. --- parquet-testing | 2 +- parquet/src/arrow/arrow_reader/mod.rs | 91 +---------- parquet/src/arrow/schema/primitive.rs | 35 +++++ parquet/src/file/metadata/options.rs | 4 +- parquet/src/file/metadata/thrift/mod.rs | 7 - parquet/src/file/serialized_reader.rs | 60 ++++++- parquet/src/schema/types.rs | 148 +++++++++++------- parquet/tests/arrow_reader/parquet_testing.rs | 66 +++++++- 8 files changed, 257 insertions(+), 156 deletions(-) diff --git a/parquet-testing b/parquet-testing index 09f3cdbde453..4b1ce4502aff 160000 --- a/parquet-testing +++ b/parquet-testing @@ -1 +1 @@ -Subproject commit 09f3cdbde45302f0f0c689c950e465e98a9df960 +Subproject commit 4b1ce4502afff8d20c9b4bb08d07e04e21cdeff3 diff --git a/parquet/src/arrow/arrow_reader/mod.rs b/parquet/src/arrow/arrow_reader/mod.rs index 5635e1812682..db43a4ae5145 100644 --- a/parquet/src/arrow/arrow_reader/mod.rs +++ b/parquet/src/arrow/arrow_reader/mod.rs @@ -830,8 +830,8 @@ impl ArrowReaderOptions { /// logical type when reading (parquet-format GH-607). /// /// Default is `false`: such combinations return an error. When `true`, the - /// column is exposed as its physical type with no logical annotation, and - /// column statistics are ignored. + /// 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); @@ -1821,7 +1821,7 @@ pub(crate) mod tests { virtual_type::{RowGroupIndex, RowNumber}, }; use crate::arrow::{ArrowWriter, ProjectionMask}; - use crate::basic::{ConvertedType, Encoding, LogicalType, Repetition, Type as PhysicalType}; + use crate::basic::{ConvertedType, Encoding, Repetition, Type as PhysicalType}; use crate::column::reader::decoder::REPETITION_LEVELS_BATCH_SIZE; use crate::data_type::{ BoolType, ByteArray, ByteArrayType, DataType, DoubleType, FixedLenByteArray, @@ -6167,89 +6167,4 @@ pub(crate) mod tests { (Bytes::from(buf), metadata) } - - fn parquet_file_with_int32_uuid_logical_type() -> Vec { - let field = Type::primitive_type_builder("id", PhysicalType::INT32) - .with_repetition(Repetition::REQUIRED) - .with_logical_type(Some(LogicalType::Uuid)) - .with_skip_logical_physical_validation(true) - .build() - .unwrap(); - let schema = Arc::new( - Type::group_type_builder("schema") - .with_fields(vec![Arc::new(field)]) - .build() - .unwrap(), - ); - let props = Arc::new( - WriterProperties::builder() - .set_statistics_enabled(EnabledStatistics::Chunk) - .build(), - ); - - let mut buf = Vec::new(); - let mut writer = SerializedFileWriter::new(&mut buf, schema, props).unwrap(); - let mut row_group_writer = writer.next_row_group().unwrap(); - let mut col_writer = row_group_writer.next_column().unwrap().unwrap(); - col_writer - .typed::() - .write_batch(&[1, 2, 3], None, None) - .unwrap(); - col_writer.close().unwrap(); - row_group_writer.close().unwrap(); - let written = writer.close().unwrap(); - assert!( - written.row_group(0).column(0).statistics().is_some(), - "test fixture must write column statistics so ignore-on-read can be asserted" - ); - buf - } - - #[test] - fn test_int32_uuid_logical_type_errors_by_default() { - let file = Bytes::from(parquet_file_with_int32_uuid_logical_type()); - let err = ParquetRecordBatchReaderBuilder::try_new(file).unwrap_err(); - assert!( - err.to_string() - .contains("Cannot annotate Uuid from INT32 for field 'id'"), - "{err}" - ); - } - - #[test] - fn test_int32_uuid_logical_type_coerced_with_option() { - let file = Bytes::from(parquet_file_with_int32_uuid_logical_type()); - let options = ArrowReaderOptions::new().with_coerce_incompatible_logical_types(true); - let builder = ParquetRecordBatchReaderBuilder::try_new_with_options(file, options).unwrap(); - - let parquet_col = builder.metadata().file_metadata().schema_descr().column(0); - assert_eq!(parquet_col.physical_type(), PhysicalType::INT32); - assert_eq!(parquet_col.logical_type_ref(), None); - assert!( - parquet_col - .get_basic_info() - .incompatible_logical_type_coerced() - ); - assert!( - builder - .metadata() - .row_group(0) - .column(0) - .statistics() - .is_none(), - "GH-607: statistics for coerced columns must be ignored" - ); - - let schema = builder.schema(); - assert_eq!(schema.field(0).data_type(), &ArrowDataType::Int32); - - let mut reader = builder.build().unwrap(); - let batch = reader.next().unwrap().unwrap(); - assert!(reader.next().is_none()); - assert_eq!(batch.num_rows(), 3); - let values = batch - .column(0) - .as_primitive::(); - assert_eq!(values.values(), &[1, 2, 3]); - } } diff --git a/parquet/src/arrow/schema/primitive.rs b/parquet/src/arrow/schema/primitive.rs index 6fcf86b66226..863bda9cb7e0 100644 --- a/parquet/src/arrow/schema/primitive.rs +++ b/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 { 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 Result { 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 + ); + } } diff --git a/parquet/src/file/metadata/options.rs b/parquet/src/file/metadata/options.rs index b677f01f8607..76f905f423b9 100644 --- a/parquet/src/file/metadata/options.rs +++ b/parquet/src/file/metadata/options.rs @@ -251,8 +251,8 @@ impl ParquetMetaDataOptions { /// be treated as an unknown logical type when reading. /// /// Default is `false`: such combinations return an error, matching historical - /// behavior. When `true`, the column is exposed as its physical type with no - /// logical annotation, and column statistics are ignored (parquet-format GH-607). + /// 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 } diff --git a/parquet/src/file/metadata/thrift/mod.rs b/parquet/src/file/metadata/thrift/mod.rs index 211789461b9c..71f6ad6d2c07 100644 --- a/parquet/src/file/metadata/thrift/mod.rs +++ b/parquet/src/file/metadata/thrift/mod.rs @@ -438,13 +438,6 @@ fn read_column_metadata( skip_col_stats = opts.skip_column_stats(col_index); skip_size_stats = opts.skip_size_stats(col_index); } - if column - .column_descr - .get_basic_info() - .incompatible_logical_type_coerced() - { - skip_col_stats = true; - } // struct ColumnMetaData { // 1: required Type type diff --git a/parquet/src/file/serialized_reader.rs b/parquet/src/file/serialized_reader.rs index b70d12690e92..21aed638d974 100644 --- a/parquet/src/file/serialized_reader.rs +++ b/parquet/src/file/serialized_reader.rs @@ -208,8 +208,8 @@ impl ReadOptionsBuilder { /// logical type when reading (parquet-format GH-607). /// /// Default is `false`: such combinations return an error. When `true`, the - /// column is exposed as its physical type with no logical annotation, and - /// column statistics are ignored. + /// 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); @@ -2975,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); + } } diff --git a/parquet/src/schema/types.rs b/parquet/src/schema/types.rs index bca369ca6938..c0337f648017 100644 --- a/parquet/src/schema/types.rs +++ b/parquet/src/schema/types.rs @@ -263,10 +263,6 @@ pub struct PrimitiveTypeBuilder<'a> { /// unknown logical type instead of returning an error. Used when reading /// files, not when constructing a schema to write. coerce_incompatible_logical_types: bool, - /// When true, skip the physical/logical compatibility check so tests can - /// emit files that other writers may produce. The logical type is kept. - #[cfg(test)] - skip_logical_physical_validation: bool, } impl<'a> PrimitiveTypeBuilder<'a> { @@ -283,8 +279,6 @@ impl<'a> PrimitiveTypeBuilder<'a> { scale: -1, id: None, coerce_incompatible_logical_types: false, - #[cfg(test)] - skip_logical_physical_validation: false, } } @@ -337,10 +331,12 @@ impl<'a> PrimitiveTypeBuilder<'a> { } /// Treat incompatible physical/logical type combinations as an unknown - /// logical type (physical type, no annotation) instead of erroring. + /// logical type instead of erroring. /// - /// This is the parquet-format GH-607 reader behavior. Do not use when - /// building a schema to write; invalid combinations should still fail. + /// The logical type is rewritten to [`LogicalType::_Unknown`] with sort + /// order [`SortOrder::UNDEFINED`]. This is the parquet-format GH-607 + /// reader behavior. Do not use when building a schema to write; invalid + /// combinations should still fail. pub(crate) fn with_coerce_incompatible_logical_types(self, value: bool) -> Self { Self { coerce_incompatible_logical_types: value, @@ -348,17 +344,6 @@ impl<'a> PrimitiveTypeBuilder<'a> { } } - /// Skip physical/logical compatibility checks, keeping the logical type. - /// - /// Only for tests that need to write a file with an invalid combination. - #[cfg(test)] - pub(crate) fn with_skip_logical_physical_validation(self, value: bool) -> Self { - Self { - skip_logical_physical_validation: value, - ..self - } - } - /// Creates a new `PrimitiveType` instance from the collected attributes. /// Returns `Err` in case of any building conditions are not met. pub fn build(self) -> Result { @@ -368,8 +353,7 @@ impl<'a> PrimitiveTypeBuilder<'a> { converted_type: self.converted_type, logical_type: self.logical_type.clone(), id: self.id, - sort_order: SortOrder::SIGNED, - incompatible_logical_type_coerced: false, + sort_order: SortOrder::UNDEFINED, }; // Check length before logical type, since it is used for logical type validation. @@ -484,22 +468,10 @@ impl<'a> PrimitiveTypeBuilder<'a> { if let Some(err) = combo_error { if self.coerce_incompatible_logical_types { - // parquet-format GH-607: treat as unknown logical type. - basic_info.logical_type = None; - basic_info.converted_type = self.converted_type; - basic_info.incompatible_logical_type_coerced = true; + // parquet-format GH-607: treat as an unknown logical type. + basic_info.logical_type = Some(LogicalType::_Unknown { field_id: 0 }); + basic_info.converted_type = ConvertedType::NONE; } else { - #[cfg(test)] - { - if self.skip_logical_physical_validation { - if self.converted_type == ConvertedType::NONE { - basic_info.converted_type = self.logical_type.clone().into(); - } - } else { - return Err(err); - } - } - #[cfg(not(test))] return Err(err); } } else if self.converted_type == ConvertedType::NONE { @@ -740,7 +712,6 @@ impl<'a> GroupTypeBuilder<'a> { logical_type: self.logical_type.clone(), id: self.id, sort_order: SortOrder::UNDEFINED, - incompatible_logical_type_coerced: false, }; // Populate the converted type if only the logical type is populated if self.logical_type.is_some() && self.converted_type == ConvertedType::NONE { @@ -853,9 +824,6 @@ pub struct BasicTypeInfo { logical_type: Option, id: Option, sort_order: SortOrder, - /// Set when an incompatible logical type was stripped while reading - /// (parquet-format GH-607). Statistics for this column should be ignored. - incompatible_logical_type_coerced: bool, } impl HeapSize for BasicTypeInfo { @@ -917,14 +885,6 @@ impl BasicTypeInfo { pub fn sort_order(&self) -> SortOrder { self.sort_order } - - /// Returns `true` if an incompatible logical type was stripped while reading - /// this field (parquet-format GH-607). - /// - /// Statistics for such columns should be ignored. - pub(crate) fn incompatible_logical_type_coerced(&self) -> bool { - self.incompatible_logical_type_coerced - } } // ---------------------------------------------------------------------- @@ -2979,16 +2939,96 @@ mod tests { ); } + fn primitive_logical_schema_elements( + name: &str, + physical: PhysicalType, + logical: LogicalType, + ) -> Vec> { + vec![ + SchemaElement { + r#type: None, + type_length: None, + repetition_type: None, + name: "schema", + num_children: Some(1), + converted_type: None, + scale: None, + precision: None, + field_id: None, + logical_type: None, + }, + SchemaElement { + r#type: Some(physical), + type_length: None, + repetition_type: Some(Repetition::REQUIRED), + name, + num_children: None, + converted_type: None, + scale: None, + precision: None, + field_id: None, + logical_type: Some(logical), + }, + ] + } + + fn assert_coerced_unknown(col: &Type, physical: PhysicalType) { + assert_eq!(col.get_physical_type(), physical); + assert_eq!( + col.get_basic_info().logical_type_ref(), + Some(&LogicalType::_Unknown { field_id: 0 }) + ); + assert_eq!(col.get_basic_info().converted_type(), ConvertedType::NONE); + // UUID would be UNSIGNED and INT32/INT64 would be SIGNED. After coerce + // the sort order must be UNDEFINED, not a leftover physical default. + assert_eq!(col.get_basic_info().sort_order(), SortOrder::UNDEFINED); + assert_eq!( + ColumnOrder::column_order_for_type( + col.get_basic_info().logical_type_ref(), + col.get_basic_info().converted_type(), + physical, + ) + .sort_order(), + SortOrder::UNDEFINED + ); + } + #[test] fn test_int32_uuid_coerced_to_physical_type() { let schema = parquet_schema_from_array_opts(int32_uuid_schema_elements(), true).unwrap(); let fields = schema.get_fields(); assert_eq!(fields.len(), 1); - let col = &fields[0]; - assert_eq!(col.get_physical_type(), PhysicalType::INT32); - assert_eq!(col.get_basic_info().logical_type_ref(), None); - assert_eq!(col.get_basic_info().converted_type(), ConvertedType::NONE); - assert!(col.get_basic_info().incompatible_logical_type_coerced()); + assert_coerced_unknown(&fields[0], PhysicalType::INT32); + } + + #[test] + fn test_int64_uuid_coerced_sort_order_is_undefined() { + let schema = parquet_schema_from_array_opts( + primitive_logical_schema_elements("id", PhysicalType::INT64, LogicalType::Uuid), + true, + ) + .unwrap(); + assert_coerced_unknown(&schema.get_fields()[0], PhysicalType::INT64); + } + + #[test] + fn test_byte_array_uuid_coerced_sort_order_is_undefined() { + let schema = parquet_schema_from_array_opts( + primitive_logical_schema_elements("id", PhysicalType::BYTE_ARRAY, LogicalType::Uuid), + true, + ) + .unwrap(); + assert_coerced_unknown(&schema.get_fields()[0], PhysicalType::BYTE_ARRAY); + } + + #[test] + fn test_float16_on_int32_coerced_to_unknown() { + let schema = parquet_schema_from_array_opts( + primitive_logical_schema_elements("id", PhysicalType::INT32, LogicalType::Float16), + true, + ) + .unwrap(); + assert_coerced_unknown(&schema.get_fields()[0], PhysicalType::INT32); } #[test] diff --git a/parquet/tests/arrow_reader/parquet_testing.rs b/parquet/tests/arrow_reader/parquet_testing.rs index ef92e6b70f0e..a01bb03802ad 100644 --- a/parquet/tests/arrow_reader/parquet_testing.rs +++ b/parquet/tests/arrow_reader/parquet_testing.rs @@ -22,9 +22,9 @@ use arrow::util::test_util::parquet_test_data; use arrow_array::cast::AsArray; use arrow_array::{Array, ArrayRef, BinaryArray, Int64Array, RecordBatch, StringArray, types}; -use arrow_schema::{ArrowError, Field, Schema, TimeUnit}; +use arrow_schema::{ArrowError, DataType, Field, Schema, TimeUnit}; use parquet::arrow::arrow_reader::{ArrowReaderOptions, ParquetRecordBatchReaderBuilder}; -use parquet::basic::{LogicalType, Type as PhysicalType}; +use parquet::basic::{LogicalType, SortOrder, Type as PhysicalType}; use std::fs::File; use std::path::PathBuf; use std::sync::Arc; @@ -285,3 +285,65 @@ fn test_json_and_bson_logical_types() { ]) ); } + +fn int32_uuid_incompatible_path() -> PathBuf { + PathBuf::from(parquet_test_data()).join("int32_with_uuid_logical_type.parquet") +} + +#[test] +fn test_int32_with_uuid_logical_type_errors_by_default() { + let file = File::open(int32_uuid_incompatible_path()).unwrap(); + let err = ParquetRecordBatchReaderBuilder::try_new(file).unwrap_err(); + assert!( + err.to_string() + .contains("Cannot annotate Uuid from INT32 for field 'int32_uuid'"), + "{err}" + ); +} + +#[test] +fn test_int32_with_uuid_logical_type_coerced_with_option() { + let file = File::open(int32_uuid_incompatible_path()).unwrap(); + let options = ArrowReaderOptions::new().with_coerce_incompatible_logical_types(true); + let builder = ParquetRecordBatchReaderBuilder::try_new_with_options(file, options).unwrap(); + + let parquet_col = builder.metadata().file_metadata().schema_descr().column(0); + assert_eq!(parquet_col.name(), "int32_uuid"); + assert_eq!(parquet_col.physical_type(), PhysicalType::INT32); + assert_eq!( + parquet_col.logical_type_ref(), + Some(&LogicalType::_Unknown { field_id: 0 }) + ); + assert_eq!( + parquet_col.get_basic_info().sort_order(), + SortOrder::UNDEFINED + ); + assert_eq!( + builder + .metadata() + .file_metadata() + .column_order(0) + .sort_order(), + SortOrder::UNDEFINED + ); + assert!( + builder + .metadata() + .row_group(0) + .column(0) + .statistics() + .is_some(), + "statistics should be retained when coercing incompatible logical types" + ); + + let schema = builder.schema(); + assert_eq!(schema.field(0).name(), "int32_uuid"); + assert_eq!(schema.field(0).data_type(), &DataType::Int32); + + let mut reader = builder.build().unwrap(); + let batch = reader.next().unwrap().unwrap(); + assert!(reader.next().is_none()); + assert_eq!(batch.num_rows(), 10); + let values = batch.column(0).as_primitive::(); + assert_eq!(values.values(), &[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]); +}