Introduce FileType reference implementation

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<Option<LogicalType>> 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,

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);
+    }
 }

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<Type> {
+        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<LogicalType>)] = &[
+        ("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<LogicalType>) -> 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<TypePtr> {
+        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 {