diff --git a/parquet/src/basic.rs b/parquet/src/basic.rs index b4f18f311723..e0fb0bda8f76 100644 --- a/parquet/src/basic.rs +++ b/parquet/src/basic.rs @@ -295,6 +295,8 @@ union LogicalType { 17: (GeometryType) Geometry /// A geospatial feature in the WKB format with an explicit (non-linear/non-planar) edges interpolation. 18: (GeographyType) Geography + /// A reference to a range of bytes, stored inline, elsewhere in this file, or in an external file. + 19: File } ); @@ -1054,6 +1056,7 @@ impl ColumnOrder { LogicalType::Variant(_) | LogicalType::Geometry(_) | LogicalType::Geography(_) + | LogicalType::File | LogicalType::_Unknown { .. } => SortOrder::UNDEFINED, }, // Fall back to converted type @@ -1242,6 +1245,7 @@ impl From> for ConvertedType { | LogicalType::Variant(_) | LogicalType::Geometry(_) | LogicalType::Geography(_) + | LogicalType::File | LogicalType::_Unknown { .. } | LogicalType::Unknown => ConvertedType::NONE, }, @@ -1341,6 +1345,7 @@ impl str::FromStr for LogicalType { )), "FLOAT16" => Ok(LogicalType::Float16), "VARIANT" => Ok(LogicalType::variant(None)), + "FILE" => Ok(LogicalType::File), "GEOMETRY" => Ok(LogicalType::geometry(None)), "GEOGRAPHY" => Ok(LogicalType::geography( None, diff --git a/parquet/src/schema/printer.rs b/parquet/src/schema/printer.rs index e43e09b30fcd..8ce07fb2fd93 100644 --- a/parquet/src/schema/printer.rs +++ b/parquet/src/schema/printer.rs @@ -337,6 +337,7 @@ fn print_logical_and_converted( format!("GEOGRAPHY({algorithm})") } } + LogicalType::File => "FILE".to_string(), LogicalType::Unknown => "UNKNOWN".to_string(), LogicalType::_Unknown { field_id } => format!("_Unknown({field_id})"), }, @@ -1173,4 +1174,43 @@ mod tests { assert_print_parse_message(message); } + + #[test] + fn test_print_file_type() { + let mut s = String::new(); + { + let mut p = Printer::new(&mut s); + let uri_field = Arc::new( + Type::primitive_type_builder("uri", PhysicalType::BYTE_ARRAY) + .with_repetition(Repetition::OPTIONAL) + .with_logical_type(Some(LogicalType::String)) + .build() + .unwrap(), + ); + let size_field = Arc::new( + Type::primitive_type_builder("size", PhysicalType::INT64) + .with_repetition(Repetition::OPTIONAL) + .build() + .unwrap(), + ); + let file_field = Type::group_type_builder("f") + .with_repetition(Repetition::REQUIRED) + .with_logical_type(Some(LogicalType::File)) + .with_fields(vec![uri_field, size_field]) + .build() + .unwrap(); + let message = Type::group_type_builder("schema") + .with_fields(vec![Arc::new(file_field)]) + .build() + .unwrap(); + p.print(&message); + } + let expected = "message schema { + REQUIRED group f (FILE) { + OPTIONAL BYTE_ARRAY uri (STRING); + OPTIONAL INT64 size; + } +}"; + assert_eq!(&mut s, expected); + } } diff --git a/parquet/src/schema/types.rs b/parquet/src/schema/types.rs index 1f9b8590fcf6..fb4616a8377e 100644 --- a/parquet/src/schema/types.rs +++ b/parquet/src/schema/types.rs @@ -349,7 +349,7 @@ impl<'a> PrimitiveTypeBuilder<'a> { } // Check that logical type and physical type are compatible match (logical_type, self.physical_type) { - (LogicalType::Map, _) | (LogicalType::List, _) => { + (LogicalType::Map, _) | (LogicalType::List, _) | (LogicalType::File, _) => { return Err(general_err!( "{:?} cannot be applied to a primitive type for field '{}'", logical_type, @@ -645,6 +645,9 @@ impl<'a> GroupTypeBuilder<'a> { /// Creates a new `GroupType` instance from the gathered attributes. pub fn build(self) -> Result { + if let Some(LogicalType::File) = &self.logical_type { + validate_file_type_fields(self.name, &self.fields)?; + } let mut basic_info = BasicTypeInfo { name: String::from(self.name), repetition: self.repetition, @@ -663,6 +666,96 @@ impl<'a> GroupTypeBuilder<'a> { } } +/// Validates the fields of a `FILE`-annotated group against the Parquet +/// [specification]. +/// +/// A `FILE` group annotates a reference to a range of bytes. Every field is +/// optional both in the schema and in the data: a writer may omit any field +/// from the group definition, and any field that is present must have a field +/// repetition type of `OPTIONAL`. The recognized fields (identified by name) +/// and their expected physical/logical types are: +/// +/// | Field | Physical type | Logical type | +/// |----------------|---------------|--------------| +/// | `uri` | `BYTE_ARRAY` | `STRING` | +/// | `offset` | `INT64` | — | +/// | `size` | `INT64` | — | +/// | `content_type` | `BYTE_ARRAY` | `STRING` | +/// | `checksum` | `BYTE_ARRAY` | `STRING` | +/// | `inline` | `BYTE_ARRAY` | — | +/// +/// [specification]: https://github.com/apache/parquet-format/blob/master/LogicalTypes.md#file +fn validate_file_type_fields(name: &str, fields: &[TypePtr]) -> Result<()> { + // (name, expected physical type, expected logical type) + const VALID_FIELDS: &[(&str, PhysicalType, Option)] = &[ + ("uri", PhysicalType::BYTE_ARRAY, Some(LogicalType::String)), + ("offset", PhysicalType::INT64, None), + ("size", PhysicalType::INT64, None), + ( + "content_type", + PhysicalType::BYTE_ARRAY, + Some(LogicalType::String), + ), + ( + "checksum", + PhysicalType::BYTE_ARRAY, + Some(LogicalType::String), + ), + ("inline", PhysicalType::BYTE_ARRAY, None), + ]; + + for field in fields { + let field_name = field.get_basic_info().name(); + let Some((_, expected_physical, expected_logical)) = + VALID_FIELDS.iter().find(|(n, _, _)| *n == field_name) + else { + return Err(general_err!( + "FILE type group '{}' contains unrecognized field '{}'. \ + Valid fields are: uri, offset, size, content_type, checksum, inline", + name, + field_name + )); + }; + + // Every field present in a FILE group must be OPTIONAL. + let is_optional = field.get_basic_info().has_repetition() + && field.get_basic_info().repetition() == Repetition::OPTIONAL; + if !is_optional { + return Err(general_err!( + "FILE type field '{}' must be OPTIONAL in group '{}'", + field_name, + name + )); + } + + // FILE fields are always primitives with a fixed physical type. + if field.is_group() { + return Err(general_err!( + "FILE type field '{}' in group '{}' must be a primitive type", + field_name, + name + )); + } + if field.get_physical_type() != *expected_physical { + return Err(general_err!( + "FILE type field '{}' in group '{}' must have physical type {:?}", + field_name, + name, + expected_physical + )); + } + if field.get_basic_info().logical_type_ref() != expected_logical.as_ref() { + return Err(general_err!( + "FILE type field '{}' in group '{}' must have logical type {:?}", + field_name, + name, + expected_logical + )); + } + } + Ok(()) +} + /// Basic type info. This contains information such as the name of the type, /// the repetition level, the logical type and the kind of the type (group, primitive). #[derive(Clone, Debug, PartialEq, Eq)] @@ -2505,6 +2598,193 @@ mod tests { assert_eq!(result_schema, expected_schema); } + /// Builds an `OPTIONAL` primitive field for use inside a `FILE` group. + fn file_field(name: &str, physical: PhysicalType, logical: Option) -> TypePtr { + Arc::new( + Type::primitive_type_builder(name, physical) + .with_repetition(Repetition::OPTIONAL) + .with_logical_type(logical) + .build() + .unwrap(), + ) + } + + /// The full set of recognized `FILE` fields, all `OPTIONAL`, per the spec. + fn all_file_fields() -> Vec { + vec![ + file_field("uri", PhysicalType::BYTE_ARRAY, Some(LogicalType::String)), + file_field("offset", PhysicalType::INT64, None), + file_field("size", PhysicalType::INT64, None), + file_field( + "content_type", + PhysicalType::BYTE_ARRAY, + Some(LogicalType::String), + ), + file_field( + "checksum", + PhysicalType::BYTE_ARRAY, + Some(LogicalType::String), + ), + file_field("inline", PhysicalType::BYTE_ARRAY, None), + ] + } + + #[test] + fn test_file_logical_type_roundtrip() { + let file_group = Arc::new( + Type::group_type_builder("f") + .with_repetition(Repetition::REQUIRED) + .with_logical_type(Some(LogicalType::File)) + .with_fields(all_file_fields()) + .build() + .unwrap(), + ); + let schema = Arc::new( + Type::group_type_builder("example") + .with_fields(vec![file_group]) + .build() + .unwrap(), + ); + let result = roundtrip_schema(schema.clone()).unwrap(); + assert_eq!(result, schema); + assert_eq!( + result.get_fields()[0].get_basic_info().logical_type_ref(), + Some(&LogicalType::File) + ); + } + + #[test] + fn test_file_logical_type_all_fields() { + let result = Type::group_type_builder("file_field") + .with_repetition(Repetition::REQUIRED) + .with_logical_type(Some(LogicalType::File)) + .with_fields(all_file_fields()) + .build(); + assert!(result.is_ok()); + assert_eq!(result.unwrap().get_fields().len(), 6); + } + + #[test] + fn test_file_logical_type_uri_only() { + // Every field is optional, so a group may define just `uri`. + let result = Type::group_type_builder("file_field") + .with_repetition(Repetition::REQUIRED) + .with_logical_type(Some(LogicalType::File)) + .with_fields(vec![file_field( + "uri", + PhysicalType::BYTE_ARRAY, + Some(LogicalType::String), + )]) + .build(); + assert!(result.is_ok()); + assert_eq!( + result.unwrap().get_basic_info().logical_type_ref(), + Some(&LogicalType::File) + ); + } + + #[test] + fn test_file_logical_type_inline_only() { + // An inline-only group need only define `inline`. + let result = Type::group_type_builder("inline_file") + .with_repetition(Repetition::REQUIRED) + .with_logical_type(Some(LogicalType::File)) + .with_fields(vec![file_field("inline", PhysicalType::BYTE_ARRAY, None)]) + .build(); + assert!(result.is_ok()); + } + + #[test] + fn test_file_logical_type_empty_group_is_allowed() { + // No field is mandatory in the schema, so an empty group is valid. + let result = Type::group_type_builder("empty_file") + .with_repetition(Repetition::REQUIRED) + .with_logical_type(Some(LogicalType::File)) + .with_fields(vec![]) + .build(); + assert!(result.is_ok()); + } + + #[test] + fn test_file_logical_type_rejects_unrecognized_field() { + let unknown_field = file_field("unknown_field", PhysicalType::BYTE_ARRAY, None); + let result = Type::group_type_builder("bad_file") + .with_repetition(Repetition::REQUIRED) + .with_logical_type(Some(LogicalType::File)) + .with_fields(vec![ + file_field("uri", PhysicalType::BYTE_ARRAY, Some(LogicalType::String)), + unknown_field, + ]) + .build(); + assert_eq!( + result.unwrap_err().to_string(), + "Parquet error: FILE type group 'bad_file' contains unrecognized field \ + 'unknown_field'. Valid fields are: uri, offset, size, content_type, \ + checksum, inline" + ); + } + + #[test] + fn test_file_logical_type_requires_optional_fields() { + // A REQUIRED field is no longer valid: every field must be OPTIONAL. + let uri_field = Arc::new( + Type::primitive_type_builder("uri", PhysicalType::BYTE_ARRAY) + .with_repetition(Repetition::REQUIRED) + .with_logical_type(Some(LogicalType::String)) + .build() + .unwrap(), + ); + let result = Type::group_type_builder("required_uri") + .with_repetition(Repetition::REQUIRED) + .with_logical_type(Some(LogicalType::File)) + .with_fields(vec![uri_field]) + .build(); + assert_eq!( + result.unwrap_err().to_string(), + "Parquet error: FILE type field 'uri' must be OPTIONAL in group 'required_uri'" + ); + } + + #[test] + fn test_file_logical_type_rejects_wrong_physical_type() { + // `size` must be an INT64, not a BYTE_ARRAY. + let bad_size = file_field("size", PhysicalType::BYTE_ARRAY, None); + let result = Type::group_type_builder("bad_size") + .with_repetition(Repetition::REQUIRED) + .with_logical_type(Some(LogicalType::File)) + .with_fields(vec![bad_size]) + .build(); + assert_eq!( + result.unwrap_err().to_string(), + "Parquet error: FILE type field 'size' in group 'bad_size' must have physical type INT64" + ); + } + + #[test] + fn test_file_logical_type_rejects_wrong_logical_type() { + // `uri` must carry the STRING logical type. + let bad_uri = file_field("uri", PhysicalType::BYTE_ARRAY, None); + let result = Type::group_type_builder("bad_uri") + .with_repetition(Repetition::REQUIRED) + .with_logical_type(Some(LogicalType::File)) + .with_fields(vec![bad_uri]) + .build(); + assert_eq!( + result.unwrap_err().to_string(), + "Parquet error: FILE type field 'uri' in group 'bad_uri' must have logical type \ + Some(String)" + ); + } + + #[test] + fn test_file_logical_type_not_allowed_on_primitive() { + let result = Type::primitive_type_builder("bad", PhysicalType::BYTE_ARRAY) + .with_repetition(Repetition::REQUIRED) + .with_logical_type(Some(LogicalType::File)) + .build(); + assert!(result.is_err()); + } + #[test] fn test_parquet_schema_from_array_rejects_negative_num_children() { let elements = vec![SchemaElement {