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 6cc07ffbf52a..db43a4ae5145 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 + /// 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. 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 3f71c5a0f4c3..76f905f423b9 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 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)] diff --git a/parquet/src/file/metadata/thrift/mod.rs b/parquet/src/file/metadata/thrift/mod.rs index 8d590079cc03..71f6ad6d2c07 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, @@ -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::(&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..21aed638d974 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 + /// 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); + } } diff --git a/parquet/src/schema/types.rs b/parquet/src/schema/types.rs index c6927fb71f84..c0337f648017 100644 --- a/parquet/src/schema/types.rs +++ b/parquet/src/schema/types.rs @@ -259,6 +259,10 @@ 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, } impl<'a> PrimitiveTypeBuilder<'a> { @@ -274,6 +278,7 @@ impl<'a> PrimitiveTypeBuilder<'a> { precision: -1, scale: -1, id: None, + coerce_incompatible_logical_types: false, } } @@ -325,22 +330,30 @@ impl<'a> PrimitiveTypeBuilder<'a> { Self { id, ..self } } + /// Treat incompatible physical/logical type combinations as an unknown + /// logical type instead of erroring. + /// + /// 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, + ..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::UNDEFINED, }; // Check length before logical type, since it is used for logical type validation. @@ -355,21 +368,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 +388,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 +408,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 +457,36 @@ 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 an unknown logical type. + basic_info.logical_type = Some(LogicalType::_Unknown { field_id: 0 }); + basic_info.converted_type = ConvertedType::NONE; + } else { + 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 => { @@ -1424,6 +1460,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 +1478,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 +1506,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 +1561,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 +1588,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 +2899,222 @@ 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}" + ); + } + + 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); + 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] + 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}" + ); + } } 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]); +}