parquet/schema/

types.rs

// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

//! Contains structs and methods to build Parquet schema and schema descriptors.

use std::vec::IntoIter;
use std::{collections::HashMap, fmt, sync::Arc};

use crate::file::metadata::HeapSize;
use crate::file::metadata::thrift::SchemaElement;

use crate::basic::{
    ColumnOrder, ConvertedType, IntType, LogicalType, Repetition, SortOrder, TimeType, TimeUnit,
    Type as PhysicalType,
};
use crate::errors::{ParquetError, Result};

// ----------------------------------------------------------------------
// Parquet Type definitions

/// Type alias for `Arc<Type>`.
pub type TypePtr = Arc<Type>;
/// Type alias for `Arc<SchemaDescriptor>`.
pub type SchemaDescPtr = Arc<SchemaDescriptor>;
/// Type alias for `Arc<ColumnDescriptor>`.
pub type ColumnDescPtr = Arc<ColumnDescriptor>;

/// Representation of a Parquet type.
///
/// Used to describe primitive leaf fields and structs, including top-level schema.
///
/// Note that the top-level schema is represented using [`Type::GroupType`] whose
/// repetition is `None`.
#[derive(Clone, Debug, PartialEq)]
pub enum Type {
    /// Represents a primitive leaf field.
    PrimitiveType {
        /// Basic information about the type.
        basic_info: BasicTypeInfo,
        /// Physical type of this primitive type.
        physical_type: PhysicalType,
        /// Length of this type.
        type_length: i32,
        /// Scale of this type.
        scale: i32,
        /// Precision of this type.
        precision: i32,
    },
    /// Represents a group of fields (similar to struct).
    GroupType {
        /// Basic information about the type.
        basic_info: BasicTypeInfo,
        /// Fields of this group type.
        fields: Vec<TypePtr>,
    },
}

impl HeapSize for Type {
    fn heap_size(&self) -> usize {
        match self {
            Type::PrimitiveType { basic_info, .. } => basic_info.heap_size(),
            Type::GroupType { basic_info, fields } => basic_info.heap_size() + fields.heap_size(),
        }
    }
}

impl Type {
    /// Creates primitive type builder with provided field name and physical type.
    pub fn primitive_type_builder(
        name: &str,
        physical_type: PhysicalType,
    ) -> PrimitiveTypeBuilder<'_> {
        PrimitiveTypeBuilder::new(name, physical_type)
    }

    /// Creates group type builder with provided column name.
    pub fn group_type_builder(name: &str) -> GroupTypeBuilder<'_> {
        GroupTypeBuilder::new(name)
    }

    /// Returns [`BasicTypeInfo`] information about the type.
    pub fn get_basic_info(&self) -> &BasicTypeInfo {
        match *self {
            Type::PrimitiveType { ref basic_info, .. } => basic_info,
            Type::GroupType { ref basic_info, .. } => basic_info,
        }
    }

    /// Returns this type's field name.
    pub fn name(&self) -> &str {
        self.get_basic_info().name()
    }

    /// Gets the fields from this group type.
    ///
    /// # Panics
    ///
    /// Panics if called on a non-group type
    // TODO: should we return `&[&Type]` here?
    pub fn get_fields(&self) -> &[TypePtr] {
        match *self {
            Type::GroupType { ref fields, .. } => &fields[..],
            Type::PrimitiveType { .. } => panic!("Cannot call get_fields() on a non-group type"),
        }
    }

    /// Gets physical type of this primitive type.
    ///
    /// # Panics
    ///
    /// Panics if called on a non-primitive type
    pub fn get_physical_type(&self) -> PhysicalType {
        match *self {
            Type::PrimitiveType {
                basic_info: _,
                physical_type,
                ..
            } => physical_type,
            Type::GroupType { .. } => {
                panic!("Cannot call get_physical_type() on a non-primitive type")
            }
        }
    }

    /// Gets precision of this primitive type.
    ///
    /// # Panics
    ///
    /// Panics if called on a non-primitive type
    pub fn get_precision(&self) -> i32 {
        match *self {
            Type::PrimitiveType { precision, .. } => precision,
            Type::GroupType { .. } => panic!("Cannot call get_precision() on non-primitive type"),
        }
    }

    /// Gets scale of this primitive type.
    ///
    /// # Panics
    ///
    /// Panics if called on a non-primitive type
    pub fn get_scale(&self) -> i32 {
        match *self {
            Type::PrimitiveType { scale, .. } => scale,
            Type::GroupType { .. } => panic!("Cannot call get_scale() on non-primitive type"),
        }
    }

    /// Checks if `sub_type` schema is part of current schema.
    /// This method can be used to check if projected columns are part of the root schema.
    pub fn check_contains(&self, sub_type: &Type) -> bool {
        // Names match, and repetitions match or not set for both
        let basic_match = self.get_basic_info().name() == sub_type.get_basic_info().name()
            && (self.is_schema() && sub_type.is_schema()
                || !self.is_schema()
                    && !sub_type.is_schema()
                    && self.get_basic_info().repetition()
                        == sub_type.get_basic_info().repetition());

        match *self {
            Type::PrimitiveType { .. } if basic_match && sub_type.is_primitive() => {
                self.get_physical_type() == sub_type.get_physical_type()
            }
            Type::GroupType { .. } if basic_match && sub_type.is_group() => {
                // build hashmap of name -> TypePtr
                let mut field_map = HashMap::new();
                for field in self.get_fields() {
                    field_map.insert(field.name(), field);
                }

                for field in sub_type.get_fields() {
                    if !field_map
                        .get(field.name())
                        .map(|tpe| tpe.check_contains(field))
                        .unwrap_or(false)
                    {
                        return false;
                    }
                }
                true
            }
            _ => false,
        }
    }

    /// Returns `true` if this type is a primitive type, `false` otherwise.
    pub fn is_primitive(&self) -> bool {
        matches!(*self, Type::PrimitiveType { .. })
    }

    /// Returns `true` if this type is a group type, `false` otherwise.
    pub fn is_group(&self) -> bool {
        matches!(*self, Type::GroupType { .. })
    }

    /// Returns `true` if this type is the top-level schema type (message type).
    pub fn is_schema(&self) -> bool {
        match *self {
            Type::GroupType { ref basic_info, .. } => !basic_info.has_repetition(),
            Type::PrimitiveType { .. } => false,
        }
    }

    /// Returns `true` if this type is repeated or optional.
    /// If this type doesn't have repetition defined, we treat it as required.
    pub fn is_optional(&self) -> bool {
        self.get_basic_info().has_repetition()
            && self.get_basic_info().repetition() != Repetition::REQUIRED
    }

    /// Returns `true` if this type is annotated as a list.
    pub(crate) fn is_list(&self) -> bool {
        if self.is_group() {
            let basic_info = self.get_basic_info();
            if let Some(logical_type) = basic_info.logical_type_ref() {
                return logical_type == &LogicalType::List;
            }
            return basic_info.converted_type() == ConvertedType::LIST;
        }
        false
    }

    /// Returns `true` if this type is a group with a single child field that is `repeated`.
    pub(crate) fn has_single_repeated_child(&self) -> bool {
        if self.is_group() {
            let children = self.get_fields();
            return children.len() == 1
                && children[0].get_basic_info().has_repetition()
                && children[0].get_basic_info().repetition() == Repetition::REPEATED;
        }
        false
    }
}

/// A builder for primitive types. All attributes are optional
/// except the name and physical type.
/// Note that if not specified explicitly, `Repetition::OPTIONAL` is used.
pub struct PrimitiveTypeBuilder<'a> {
    name: &'a str,
    repetition: Repetition,
    physical_type: PhysicalType,
    converted_type: ConvertedType,
    logical_type: Option<LogicalType>,
    length: i32,
    precision: i32,
    scale: i32,
    id: Option<i32>,
}

impl<'a> PrimitiveTypeBuilder<'a> {
    /// Creates new primitive type builder with provided field name and physical type.
    pub fn new(name: &'a str, physical_type: PhysicalType) -> Self {
        Self {
            name,
            repetition: Repetition::OPTIONAL,
            physical_type,
            converted_type: ConvertedType::NONE,
            logical_type: None,
            length: -1,
            precision: -1,
            scale: -1,
            id: None,
        }
    }

    /// Sets [`Repetition`] for this field and returns itself.
    pub fn with_repetition(self, repetition: Repetition) -> Self {
        Self { repetition, ..self }
    }

    /// Sets [`ConvertedType`] for this field and returns itself.
    pub fn with_converted_type(self, converted_type: ConvertedType) -> Self {
        Self {
            converted_type,
            ..self
        }
    }

    /// Sets [`LogicalType`] for this field and returns itself.
    /// If only the logical type is populated for a primitive type, the converted type
    /// will be automatically populated, and can thus be omitted.
    pub fn with_logical_type(self, logical_type: Option<LogicalType>) -> Self {
        Self {
            logical_type,
            ..self
        }
    }

    /// Sets type length and returns itself.
    /// This is only applied to FIXED_LEN_BYTE_ARRAY and INT96 (INTERVAL) types, because
    /// they maintain fixed size underlying byte array.
    /// By default, value is `0`.
    pub fn with_length(self, length: i32) -> Self {
        Self { length, ..self }
    }

    /// Sets precision for Parquet DECIMAL physical type and returns itself.
    /// By default, it equals to `0` and used only for decimal context.
    pub fn with_precision(self, precision: i32) -> Self {
        Self { precision, ..self }
    }

    /// Sets scale for Parquet DECIMAL physical type and returns itself.
    /// By default, it equals to `0` and used only for decimal context.
    pub fn with_scale(self, scale: i32) -> Self {
        Self { scale, ..self }
    }

    /// Sets optional field id and returns itself.
    pub fn with_id(self, id: Option<i32>) -> Self {
        Self { id, ..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<Type> {
        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,
        };

        // Check length before logical type, since it is used for logical type validation.
        if self.physical_type == PhysicalType::FIXED_LEN_BYTE_ARRAY && self.length < 0 {
            return Err(general_err!(
                "Invalid FIXED_LEN_BYTE_ARRAY length: {} for field '{}'",
                self.length,
                self.name
            ));
        }

        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();
            }
            // Check that logical type and physical type are compatible
            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 '{}'",
                        logical_type,
                        self.name
                    ));
                }
                (LogicalType::Enum, PhysicalType::BYTE_ARRAY) => {}
                (LogicalType::Decimal(decimal), _) => {
                    // Check that scale and precision are consistent with legacy values
                    if decimal.scale != self.scale {
                        return Err(general_err!(
                            "DECIMAL logical type scale {} must match self.scale {} for field '{}'",
                            decimal.scale,
                            self.scale,
                            self.name
                        ));
                    }
                    if decimal.precision != self.precision {
                        return Err(general_err!(
                            "DECIMAL logical type precision {} must match self.precision {} for field '{}'",
                            decimal.precision,
                            self.precision,
                            self.name
                        ));
                    }
                    self.check_decimal_precision_scale()?;
                }
                (LogicalType::Date, PhysicalType::INT32) => {}
                (
                    LogicalType::Time(TimeType {
                        unit: TimeUnit::MILLIS,
                        ..
                    }),
                    PhysicalType::INT32,
                ) => {}
                (LogicalType::Time(time), PhysicalType::INT64) => {
                    if time.unit == TimeUnit::MILLIS {
                        return Err(general_err!(
                            "Cannot use millisecond unit on INT64 type for field '{}'",
                            self.name
                        ));
                    }
                }
                (LogicalType::Timestamp(_), PhysicalType::INT64) => {}
                (LogicalType::Integer(int), PhysicalType::INT32) if int.bit_width <= 32 => {}
                (LogicalType::Integer(int), PhysicalType::INT64) if int.bit_width == 64 => {}
                // 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::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) => {
                    return Err(general_err!(
                        "FLOAT16 cannot annotate field '{}' because it is not a FIXED_LEN_BYTE_ARRAY(2) field",
                        self.name
                    ));
                }
                // 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
                    ));
                }
            }
        }

        match self.converted_type {
            ConvertedType::NONE => {}
            ConvertedType::UTF8 | ConvertedType::BSON | ConvertedType::JSON => {
                if self.physical_type != PhysicalType::BYTE_ARRAY {
                    return Err(general_err!(
                        "{} cannot annotate field '{}' because it is not a BYTE_ARRAY field",
                        self.converted_type,
                        self.name
                    ));
                }
            }
            ConvertedType::DECIMAL => {
                self.check_decimal_precision_scale()?;
            }
            ConvertedType::DATE
            | ConvertedType::TIME_MILLIS
            | ConvertedType::UINT_8
            | ConvertedType::UINT_16
            | ConvertedType::UINT_32
            | ConvertedType::INT_8
            | ConvertedType::INT_16
            | ConvertedType::INT_32 => {
                if self.physical_type != PhysicalType::INT32 {
                    return Err(general_err!(
                        "{} cannot annotate field '{}' because it is not a INT32 field",
                        self.converted_type,
                        self.name
                    ));
                }
            }
            ConvertedType::TIME_MICROS
            | ConvertedType::TIMESTAMP_MILLIS
            | ConvertedType::TIMESTAMP_MICROS
            | ConvertedType::UINT_64
            | ConvertedType::INT_64 => {
                if self.physical_type != PhysicalType::INT64 {
                    return Err(general_err!(
                        "{} cannot annotate field '{}' because it is not a INT64 field",
                        self.converted_type,
                        self.name
                    ));
                }
            }
            ConvertedType::INTERVAL => {
                if self.physical_type != PhysicalType::FIXED_LEN_BYTE_ARRAY || self.length != 12 {
                    return Err(general_err!(
                        "INTERVAL cannot annotate field '{}' because it is not a FIXED_LEN_BYTE_ARRAY(12) field",
                        self.name
                    ));
                }
            }
            ConvertedType::ENUM => {
                if self.physical_type != PhysicalType::BYTE_ARRAY {
                    return Err(general_err!(
                        "ENUM cannot annotate field '{}' because it is not a BYTE_ARRAY field",
                        self.name
                    ));
                }
            }
            _ => {
                return Err(general_err!(
                    "{} cannot be applied to primitive field '{}'",
                    self.converted_type,
                    self.name
                ));
            }
        }

        Ok(Type::PrimitiveType {
            basic_info,
            physical_type: self.physical_type,
            type_length: self.length,
            scale: self.scale,
            precision: self.precision,
        })
    }

    #[inline]
    fn check_decimal_precision_scale(&self) -> Result<()> {
        match self.physical_type {
            PhysicalType::INT32
            | PhysicalType::INT64
            | PhysicalType::BYTE_ARRAY
            | PhysicalType::FIXED_LEN_BYTE_ARRAY => (),
            _ => {
                return Err(general_err!(
                    "DECIMAL can only annotate INT32, INT64, BYTE_ARRAY and FIXED_LEN_BYTE_ARRAY"
                ));
            }
        }

        // Precision is required and must be a non-zero positive integer.
        if self.precision < 1 {
            return Err(general_err!(
                "Invalid DECIMAL precision: {}",
                self.precision
            ));
        }

        // Scale must be zero or a positive integer less than the precision.
        if self.scale < 0 {
            return Err(general_err!("Invalid DECIMAL scale: {}", self.scale));
        }

        if self.scale > self.precision {
            return Err(general_err!(
                "Invalid DECIMAL: scale ({}) cannot be greater than precision \
             ({})",
                self.scale,
                self.precision
            ));
        }

        // Check precision and scale based on physical type limitations.
        match self.physical_type {
            PhysicalType::INT32 => {
                if self.precision > 9 {
                    return Err(general_err!(
                        "Cannot represent INT32 as DECIMAL with precision {}",
                        self.precision
                    ));
                }
            }
            PhysicalType::INT64 => {
                if self.precision > 18 {
                    return Err(general_err!(
                        "Cannot represent INT64 as DECIMAL with precision {}",
                        self.precision
                    ));
                }
            }
            PhysicalType::FIXED_LEN_BYTE_ARRAY => {
                let length = self
                    .length
                    .checked_mul(8)
                    .ok_or(general_err!("Invalid length {} for Decimal", self.length))?;
                let max_precision = (2f64.powi(length - 1) - 1f64).log10().floor() as i32;

                if self.precision > max_precision {
                    return Err(general_err!(
                        "Cannot represent FIXED_LEN_BYTE_ARRAY as DECIMAL with length {} and \
                        precision {}. The max precision can only be {}",
                        self.length,
                        self.precision,
                        max_precision
                    ));
                }
            }
            _ => (), // For BYTE_ARRAY precision is not limited
        }

        Ok(())
    }
}

/// A builder for group types. All attributes are optional except the name.
/// Note that if not specified explicitly, `None` is used as the repetition of the group,
/// which means it is a root (message) type.
pub struct GroupTypeBuilder<'a> {
    name: &'a str,
    repetition: Option<Repetition>,
    converted_type: ConvertedType,
    logical_type: Option<LogicalType>,
    fields: Vec<TypePtr>,
    id: Option<i32>,
}

impl<'a> GroupTypeBuilder<'a> {
    /// Creates new group type builder with provided field name.
    pub fn new(name: &'a str) -> Self {
        Self {
            name,
            repetition: None,
            converted_type: ConvertedType::NONE,
            logical_type: None,
            fields: Vec::new(),
            id: None,
        }
    }

    /// Sets [`Repetition`] for this field and returns itself.
    pub fn with_repetition(mut self, repetition: Repetition) -> Self {
        self.repetition = Some(repetition);
        self
    }

    /// Sets [`ConvertedType`] for this field and returns itself.
    pub fn with_converted_type(self, converted_type: ConvertedType) -> Self {
        Self {
            converted_type,
            ..self
        }
    }

    /// Sets [`LogicalType`] for this field and returns itself.
    pub fn with_logical_type(self, logical_type: Option<LogicalType>) -> Self {
        Self {
            logical_type,
            ..self
        }
    }

    /// Sets a list of fields that should be child nodes of this field.
    /// Returns updated self.
    pub fn with_fields(self, fields: Vec<TypePtr>) -> Self {
        Self { fields, ..self }
    }

    /// Sets optional field id and returns itself.
    pub fn with_id(self, id: Option<i32>) -> Self {
        Self { id, ..self }
    }

    /// Creates a new `GroupType` instance from the gathered attributes.
    pub fn build(self) -> Result<Type> {
        if matches!(&self.logical_type, Some(LogicalType::File)) {
            validate_file_type_fields(self.name, &self.fields)?;
        }
        let mut basic_info = BasicTypeInfo {
            name: String::from(self.name),
            repetition: self.repetition,
            converted_type: self.converted_type,
            logical_type: self.logical_type.clone(),
            id: self.id,
            sort_order: SortOrder::UNDEFINED,
        };
        // Populate the converted type if only the logical type is populated
        if self.logical_type.is_some() && self.converted_type == ConvertedType::NONE {
            basic_info.converted_type = self.logical_type.into();
        }
        Ok(Type::GroupType {
            basic_info,
            fields: self.fields,
        })
    }
}

/// 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)]
pub struct BasicTypeInfo {
    name: String,
    repetition: Option<Repetition>,
    converted_type: ConvertedType,
    logical_type: Option<LogicalType>,
    id: Option<i32>,
    sort_order: SortOrder,
}

impl HeapSize for BasicTypeInfo {
    fn heap_size(&self) -> usize {
        // no heap allocations in any other subfield
        self.name.heap_size()
    }
}

impl BasicTypeInfo {
    /// Returns field name.
    pub fn name(&self) -> &str {
        &self.name
    }

    /// Returns `true` if type has repetition field set, `false` otherwise.
    /// This is mostly applied to group type, because primitive type always has
    /// repetition set.
    pub fn has_repetition(&self) -> bool {
        self.repetition.is_some()
    }

    /// Returns [`Repetition`] value for the type.
    ///
    /// # Panics
    ///
    /// Panics if the repetition is not set, see [`Self::has_repetition`]
    pub fn repetition(&self) -> Repetition {
        assert!(self.repetition.is_some());
        self.repetition.unwrap()
    }

    /// Returns [`ConvertedType`] value for the type.
    pub fn converted_type(&self) -> ConvertedType {
        self.converted_type
    }

    /// Return a reference to the [`LogicalType`] value for the type.
    pub fn logical_type_ref(&self) -> Option<&LogicalType> {
        self.logical_type.as_ref()
    }

    /// Returns `true` if id is set, `false` otherwise.
    pub fn has_id(&self) -> bool {
        self.id.is_some()
    }

    /// Returns id value for the type.
    ///
    /// # Panics
    ///
    /// Panics if the id is not set, see [`Self::has_id`]
    pub fn id(&self) -> i32 {
        assert!(self.id.is_some());
        self.id.unwrap()
    }

    /// Returns [`SortOrder`] for the type.
    pub fn sort_order(&self) -> SortOrder {
        self.sort_order
    }
}

// ----------------------------------------------------------------------
// Parquet descriptor definitions

/// Represents the location of a column in a Parquet schema
///
/// # Example: refer to column named `'my_column'`
/// ```
/// # use parquet::schema::types::ColumnPath;
/// let column_path = ColumnPath::from("my_column");
/// ```
///
/// # Example: refer to column named `c` in a nested struct `{a: {b: {c: ...}}}`
/// ```
/// # use parquet::schema::types::ColumnPath;
/// // form path 'a.b.c'
/// let column_path = ColumnPath::from(vec![
///   String::from("a"),
///   String::from("b"),
///   String::from("c")
/// ]);
/// ```
#[derive(Clone, PartialEq, Debug, Eq, Hash)]
pub struct ColumnPath {
    parts: Vec<String>,
}

impl HeapSize for ColumnPath {
    fn heap_size(&self) -> usize {
        self.parts.heap_size()
    }
}

impl ColumnPath {
    /// Creates new column path from vector of field names.
    pub fn new(parts: Vec<String>) -> Self {
        ColumnPath { parts }
    }

    /// Returns string representation of this column path.
    /// ```rust
    /// use parquet::schema::types::ColumnPath;
    ///
    /// let path = ColumnPath::new(vec!["a".to_string(), "b".to_string(), "c".to_string()]);
    /// assert_eq!(&path.string(), "a.b.c");
    /// ```
    pub fn string(&self) -> String {
        self.parts.join(".")
    }

    /// Appends more components to end of column path.
    /// ```rust
    /// use parquet::schema::types::ColumnPath;
    ///
    /// let mut path = ColumnPath::new(vec!["a".to_string(), "b".to_string(), "c"
    /// .to_string()]);
    /// assert_eq!(&path.string(), "a.b.c");
    ///
    /// path.append(vec!["d".to_string(), "e".to_string()]);
    /// assert_eq!(&path.string(), "a.b.c.d.e");
    /// ```
    pub fn append(&mut self, mut tail: Vec<String>) {
        self.parts.append(&mut tail);
    }

    /// Returns a slice of path components.
    pub fn parts(&self) -> &[String] {
        &self.parts
    }
}

impl fmt::Display for ColumnPath {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        write!(f, "{:?}", self.string())
    }
}

impl From<Vec<String>> for ColumnPath {
    fn from(parts: Vec<String>) -> Self {
        ColumnPath { parts }
    }
}

impl From<&str> for ColumnPath {
    fn from(single_path: &str) -> Self {
        let s = String::from(single_path);
        ColumnPath::from(s)
    }
}

impl From<String> for ColumnPath {
    fn from(single_path: String) -> Self {
        let v = vec![single_path];
        ColumnPath { parts: v }
    }
}

impl AsRef<[String]> for ColumnPath {
    fn as_ref(&self) -> &[String] {
        &self.parts
    }
}

/// Physical type for leaf-level primitive columns.
///
/// Also includes the maximum definition and repetition levels required to
/// re-assemble nested data.
#[derive(Debug, PartialEq)]
pub struct ColumnDescriptor {
    /// The "leaf" primitive type of this column
    primitive_type: TypePtr,

    /// The maximum definition level for this column
    max_def_level: i16,

    /// The maximum repetition level for this column
    max_rep_level: i16,

    /// The definition level at the nearest REPEATED ancestor, or 0 if none.
    repeated_ancestor_def_level: i16,

    /// The path of this column. For instance, "a.b.c.d".
    path: ColumnPath,
}

impl HeapSize for ColumnDescriptor {
    fn heap_size(&self) -> usize {
        // Don't include the heap size of primitive_type, this is already
        // accounted for via SchemaDescriptor::schema
        self.path.heap_size()
    }
}

impl ColumnDescriptor {
    /// Creates new descriptor for leaf-level column.
    pub fn new(
        primitive_type: TypePtr,
        max_def_level: i16,
        max_rep_level: i16,
        path: ColumnPath,
    ) -> Self {
        Self::new_with_repeated_ancestor(primitive_type, max_def_level, max_rep_level, path, 0)
    }

    pub(crate) fn new_with_repeated_ancestor(
        primitive_type: TypePtr,
        max_def_level: i16,
        max_rep_level: i16,
        path: ColumnPath,
        repeated_ancestor_def_level: i16,
    ) -> Self {
        Self {
            primitive_type,
            max_def_level,
            max_rep_level,
            repeated_ancestor_def_level,
            path,
        }
    }

    /// Returns maximum definition level for this column.
    #[inline]
    pub fn max_def_level(&self) -> i16 {
        self.max_def_level
    }

    /// Returns maximum repetition level for this column.
    #[inline]
    pub fn max_rep_level(&self) -> i16 {
        self.max_rep_level
    }

    /// Returns the definition level at the nearest REPEATED ancestor, or 0 if none.
    #[inline]
    pub fn repeated_ancestor_def_level(&self) -> i16 {
        self.repeated_ancestor_def_level
    }

    /// Returns [`ColumnPath`] for this column.
    pub fn path(&self) -> &ColumnPath {
        &self.path
    }

    /// Returns self type [`Type`] for this leaf column.
    pub fn self_type(&self) -> &Type {
        self.primitive_type.as_ref()
    }

    /// Returns self type [`TypePtr`]  for this leaf
    /// column.
    pub fn self_type_ptr(&self) -> TypePtr {
        self.primitive_type.clone()
    }

    /// Returns [`BasicTypeInfo`] information for this leaf column.
    pub fn get_basic_info(&self) -> &BasicTypeInfo {
        self.primitive_type.get_basic_info()
    }

    /// Returns column name.
    pub fn name(&self) -> &str {
        self.primitive_type.name()
    }

    /// Returns [`ConvertedType`] for this column.
    pub fn converted_type(&self) -> ConvertedType {
        self.primitive_type.get_basic_info().converted_type()
    }

    /// Returns a reference to the [`LogicalType`] for this column.
    pub fn logical_type_ref(&self) -> Option<&LogicalType> {
        self.primitive_type.get_basic_info().logical_type_ref()
    }

    /// Returns physical type for this column.
    ///
    /// # Panics
    ///
    /// Panics if called on a non-primitive type
    pub fn physical_type(&self) -> PhysicalType {
        match self.primitive_type.as_ref() {
            Type::PrimitiveType { physical_type, .. } => *physical_type,
            Type::GroupType { .. } => panic!("Expected primitive type!"),
        }
    }

    /// Returns type length for this column.
    ///
    /// # Panics
    ///
    /// Panics if called on a non-primitive type
    pub fn type_length(&self) -> i32 {
        match self.primitive_type.as_ref() {
            Type::PrimitiveType { type_length, .. } => *type_length,
            Type::GroupType { .. } => panic!("Expected primitive type!"),
        }
    }

    /// Returns type precision for this column.
    ///
    /// # Panics
    ///
    /// Panics if called on a non-primitive type
    pub fn type_precision(&self) -> i32 {
        match self.primitive_type.as_ref() {
            Type::PrimitiveType { precision, .. } => *precision,
            Type::GroupType { .. } => panic!("Expected primitive type!"),
        }
    }

    /// Returns type scale for this column.
    ///
    /// # Panics
    ///
    /// Panics if called on a non-primitive type
    pub fn type_scale(&self) -> i32 {
        match self.primitive_type.as_ref() {
            Type::PrimitiveType { scale, .. } => *scale,
            Type::GroupType { .. } => panic!("Expected primitive type!"),
        }
    }

    /// Returns the sort order for this column as currently defined for the logical or
    /// physical type.
    ///
    /// Returns `SortOrder::UNDEFINED` for non-primitive types.
    pub fn sort_order(&self) -> SortOrder {
        self.primitive_type.get_basic_info().sort_order()
    }
}

/// Schema of a Parquet file.
///
/// Encapsulates the file's schema ([`Type`]) and [`ColumnDescriptor`]s for
/// each primitive (leaf) column.
///
/// # Example
/// ```
/// # use std::sync::Arc;
/// use parquet::schema::types::{SchemaDescriptor, Type};
/// use parquet::basic; // note there are two `Type`s that are different
/// // Schema for a table with two columns: "a" (int64) and "b" (int32, stored as a date)
/// let descriptor = SchemaDescriptor::new(
///   Arc::new(
///     Type::group_type_builder("my_schema")
///       .with_fields(vec![
///         Arc::new(
///          Type::primitive_type_builder("a", basic::Type::INT64)
///           .build().unwrap()
///         ),
///         Arc::new(
///          Type::primitive_type_builder("b", basic::Type::INT32)
///           .with_converted_type(basic::ConvertedType::DATE)
///           .with_logical_type(Some(basic::LogicalType::Date))
///           .build().unwrap()
///         ),
///      ])
///      .build().unwrap()
///   )
/// );
/// ```
#[derive(PartialEq, Clone)]
pub struct SchemaDescriptor {
    /// The top-level logical schema (the "message" type).
    ///
    /// This must be a [`Type::GroupType`] where each field is a root
    /// column type in the schema.
    schema: TypePtr,

    /// The descriptors for the physical type of each leaf column in this schema
    ///
    /// Constructed from `schema` in DFS order.
    leaves: Vec<ColumnDescPtr>,

    /// Mapping from a leaf column's index to the root column index that it
    /// comes from.
    ///
    /// For instance: the leaf `a.b.c.d` would have a link back to `a`:
    /// ```text
    /// -- a  <-----+
    /// -- -- b     |
    /// -- -- -- c  |
    /// -- -- -- -- d
    /// ```
    leaf_to_base: Vec<usize>,
}

impl fmt::Debug for SchemaDescriptor {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        // Skip leaves and leaf_to_base as they only a cache information already found in `schema`
        f.debug_struct("SchemaDescriptor")
            .field("schema", &self.schema)
            .finish()
    }
}

// Need to implement HeapSize in this module as the fields are private
impl HeapSize for SchemaDescriptor {
    fn heap_size(&self) -> usize {
        self.schema.heap_size() + self.leaves.heap_size() + self.leaf_to_base.heap_size()
    }
}

impl SchemaDescriptor {
    /// Creates new schema descriptor from Parquet schema.
    pub fn new(tp: TypePtr) -> Self {
        const INIT_SCHEMA_DEPTH: usize = 16;
        assert!(tp.is_group(), "SchemaDescriptor should take a GroupType");
        // unwrap should be safe since we just asserted tp is a group
        let n_leaves = num_leaves(&tp).unwrap();
        let mut leaves = Vec::with_capacity(n_leaves);
        let mut leaf_to_base = Vec::with_capacity(n_leaves);
        let mut path = Vec::with_capacity(INIT_SCHEMA_DEPTH);
        for (root_idx, f) in tp.get_fields().iter().enumerate() {
            path.clear();
            build_tree(
                f,
                root_idx,
                0,
                0,
                0,
                &mut leaves,
                &mut leaf_to_base,
                &mut path,
            );
        }

        Self {
            schema: tp,
            leaves,
            leaf_to_base,
        }
    }

    /// Returns [`ColumnDescriptor`] for a field position.
    ///
    /// # Panics
    ///
    /// Panics if `i >= self.num_columns()`
    pub fn column(&self, i: usize) -> ColumnDescPtr {
        assert!(
            i < self.leaves.len(),
            "Index out of bound: {} not in [0, {})",
            i,
            self.leaves.len()
        );
        self.leaves[i].clone()
    }

    /// Returns slice of [`ColumnDescriptor`].
    pub fn columns(&self) -> &[ColumnDescPtr] {
        &self.leaves
    }

    /// Returns number of leaf-level columns.
    pub fn num_columns(&self) -> usize {
        self.leaves.len()
    }

    /// Returns column root [`Type`] for a leaf position.
    pub fn get_column_root(&self, i: usize) -> &Type {
        let result = self.column_root_of(i);
        result.as_ref()
    }

    /// Returns column root [`Type`] pointer for a leaf position.
    ///
    /// # Panics
    ///
    /// Panics if `i >= self.num_columns()`
    pub fn get_column_root_ptr(&self, i: usize) -> TypePtr {
        let result = self.column_root_of(i);
        result.clone()
    }

    /// Returns the index of the root column for a field position
    ///
    /// # Panics
    ///
    /// Panics if `leaf` is out of bounds
    pub fn get_column_root_idx(&self, leaf: usize) -> usize {
        assert!(
            leaf < self.leaves.len(),
            "Index out of bound: {} not in [0, {})",
            leaf,
            self.leaves.len()
        );

        *self
            .leaf_to_base
            .get(leaf)
            .unwrap_or_else(|| panic!("Expected a value for index {leaf} but found None"))
    }

    fn column_root_of(&self, i: usize) -> &TypePtr {
        &self.schema.get_fields()[self.get_column_root_idx(i)]
    }

    /// Returns schema as [`Type`].
    pub fn root_schema(&self) -> &Type {
        self.schema.as_ref()
    }

    /// Returns schema as [`TypePtr`] for cheap cloning.
    pub fn root_schema_ptr(&self) -> TypePtr {
        self.schema.clone()
    }

    /// Returns schema name.
    pub fn name(&self) -> &str {
        self.schema.name()
    }
}

// walk tree and count nodes
pub(crate) fn num_nodes(tp: &TypePtr) -> Result<usize> {
    if !tp.is_group() {
        return Err(general_err!("Root schema must be Group type"));
    }
    let mut n_nodes = 1usize; // count root
    for f in tp.get_fields() {
        count_nodes(f, &mut n_nodes);
    }
    Ok(n_nodes)
}

pub(crate) fn count_nodes(tp: &TypePtr, n_nodes: &mut usize) {
    *n_nodes += 1;
    if let Type::GroupType { fields, .. } = tp.as_ref() {
        for f in fields {
            count_nodes(f, n_nodes);
        }
    }
}

// do a quick walk of the tree to get proper sizing for SchemaDescriptor arrays
fn num_leaves(tp: &TypePtr) -> Result<usize> {
    if !tp.is_group() {
        return Err(general_err!("Root schema must be Group type"));
    }
    let mut n_leaves = 0usize;
    for f in tp.get_fields() {
        count_leaves(f, &mut n_leaves);
    }
    Ok(n_leaves)
}

fn count_leaves(tp: &TypePtr, n_leaves: &mut usize) {
    match tp.as_ref() {
        Type::PrimitiveType { .. } => *n_leaves += 1,
        Type::GroupType { fields, .. } => {
            for f in fields {
                count_leaves(f, n_leaves);
            }
        }
    }
}

#[expect(clippy::too_many_arguments)]
fn build_tree<'a>(
    tp: &'a TypePtr,
    root_idx: usize,
    mut max_rep_level: i16,
    mut max_def_level: i16,
    mut repeated_ancestor_def_level: i16,
    leaves: &mut Vec<ColumnDescPtr>,
    leaf_to_base: &mut Vec<usize>,
    path_so_far: &mut Vec<&'a str>,
) {
    assert!(tp.get_basic_info().has_repetition());

    path_so_far.push(tp.name());
    match tp.get_basic_info().repetition() {
        Repetition::OPTIONAL => {
            max_def_level += 1;
        }
        Repetition::REPEATED => {
            max_def_level += 1;
            max_rep_level += 1;
            repeated_ancestor_def_level = max_def_level;
        }
        Repetition::REQUIRED => {}
    }

    match tp.as_ref() {
        Type::PrimitiveType { .. } => {
            let mut path: Vec<String> = vec![];
            path.extend(path_so_far.iter().copied().map(String::from));
            let desc = ColumnDescriptor::new_with_repeated_ancestor(
                tp.clone(),
                max_def_level,
                max_rep_level,
                ColumnPath::new(path),
                repeated_ancestor_def_level,
            );
            leaves.push(Arc::new(desc));
            leaf_to_base.push(root_idx);
        }
        Type::GroupType { fields, .. } => {
            for f in fields {
                build_tree(
                    f,
                    root_idx,
                    max_rep_level,
                    max_def_level,
                    repeated_ancestor_def_level,
                    leaves,
                    leaf_to_base,
                    path_so_far,
                );
                path_so_far.pop();
            }
        }
    }
}

/// Checks if the logical type is valid.
fn check_logical_type(logical_type: Option<&LogicalType>) -> Result<()> {
    if let Some(LogicalType::Integer(IntType { bit_width, .. })) = logical_type
        && *bit_width != 8
        && *bit_width != 16
        && *bit_width != 32
        && *bit_width != 64
    {
        return Err(general_err!(
            "Bit width must be 8, 16, 32, or 64 for Integer logical type"
        ));
    }
    Ok(())
}

// 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<SchemaElement<'_>>) -> Result<TypePtr> {
    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

    // turn into iterator so we can take ownership of elements of the vector
    let mut elements = elements.into_iter();

    while index < num_elements {
        let t = schema_from_array_helper(&mut elements, num_elements, index)?;
        index = t.0;
        schema_nodes.push(t.1);
    }
    if schema_nodes.len() != 1 {
        return Err(general_err!(
            "Expected exactly one root node, but found {}",
            schema_nodes.len()
        ));
    }

    if !schema_nodes[0].is_group() {
        return Err(general_err!("Expected root node to be a group type"));
    }

    Ok(schema_nodes.remove(0))
}

// recursive helper function for schema conversion
fn schema_from_array_helper(
    elements: &mut IntoIter<SchemaElement<'_>>,
    num_elements: usize,
    index: usize,
) -> Result<(usize, TypePtr)> {
    // Whether or not the current node is root (message type).
    // There is only one message type node in the schema tree.
    let is_root_node = index == 0;

    if index >= num_elements {
        return Err(general_err!(
            "Index out of bound, index = {}, len = {}",
            index,
            num_elements
        ));
    }
    let element = elements.next().expect("schema vector should not be empty");

    // Check for empty schema
    if let (true, None | Some(0)) = (is_root_node, element.num_children) {
        let builder = Type::group_type_builder(element.name);
        return Ok((index + 1, Arc::new(builder.build().unwrap())));
    }

    let converted_type = element.converted_type.unwrap_or(ConvertedType::NONE);

    // LogicalType is prefered to ConvertedType, but both may be present.
    let logical_type = element.logical_type;

    check_logical_type(logical_type.as_ref())?;

    let field_id = element.field_id;
    match element.num_children {
        // From parquet-format:
        //   The children count is used to construct the nested relationship.
        //   This field is not set when the element is a primitive type
        // Sometimes parquet-cpp sets num_children field to 0 for primitive types, so we
        // have to handle this case too.
        None | Some(0) => {
            // primitive type
            if element.repetition_type.is_none() {
                return Err(general_err!(
                    "Repetition level must be defined for a primitive type"
                ));
            }
            let repetition = element.repetition_type.unwrap();
            if let Some(physical_type) = element.r#type {
                let length = element.type_length.unwrap_or(-1);
                let scale = element.scale.unwrap_or(-1);
                let precision = element.precision.unwrap_or(-1);
                let name = element.name;
                let builder = Type::primitive_type_builder(name, physical_type)
                    .with_repetition(repetition)
                    .with_converted_type(converted_type)
                    .with_logical_type(logical_type)
                    .with_length(length)
                    .with_precision(precision)
                    .with_scale(scale)
                    .with_id(field_id);
                Ok((index + 1, Arc::new(builder.build()?)))
            } else {
                let mut builder = Type::group_type_builder(element.name)
                    .with_converted_type(converted_type)
                    .with_logical_type(logical_type)
                    .with_id(field_id);
                if !is_root_node {
                    // Sometimes parquet-cpp and parquet-mr set repetition level REQUIRED or
                    // REPEATED for root node.
                    //
                    // We only set repetition for group types that are not top-level message
                    // type. According to parquet-format:
                    //   Root of the schema does not have a repetition_type.
                    //   All other types must have one.
                    builder = builder.with_repetition(repetition);
                }
                Ok((index + 1, Arc::new(builder.build().unwrap())))
            }
        }
        Some(n) => {
            let repetition = element.repetition_type;

            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)?;
                next_index = child_result.0;
                fields.push(child_result.1);
            }

            let mut builder = Type::group_type_builder(element.name)
                .with_converted_type(converted_type)
                .with_logical_type(logical_type)
                .with_fields(fields)
                .with_id(field_id);

            // Sometimes parquet-cpp and parquet-mr set repetition level REQUIRED or
            // REPEATED for root node.
            //
            // We only set repetition for group types that are not top-level message
            // type. According to parquet-format:
            //   Root of the schema does not have a repetition_type.
            //   All other types must have one.
            if !is_root_node {
                let Some(rep) = repetition else {
                    return Err(general_err!(
                        "Repetition level must be defined for non-root types"
                    ));
                };
                builder = builder.with_repetition(rep);
            }
            Ok((next_index, Arc::new(builder.build()?)))
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    use crate::{
        file::metadata::thrift::tests::{buf_to_schema_list, roundtrip_schema, schema_to_buf},
        schema::parser::parse_message_type,
    };

    // TODO: add tests for v2 types

    #[test]
    fn test_primitive_type() {
        let mut result = Type::primitive_type_builder("foo", PhysicalType::INT32)
            .with_logical_type(Some(LogicalType::integer(32, true)))
            .with_id(Some(0))
            .build();
        assert!(result.is_ok());

        if let Ok(tp) = result {
            assert!(tp.is_primitive());
            assert!(!tp.is_group());
            let basic_info = tp.get_basic_info();
            assert_eq!(basic_info.repetition(), Repetition::OPTIONAL);
            assert_eq!(
                basic_info.logical_type_ref(),
                Some(&LogicalType::integer(32, true))
            );
            assert_eq!(basic_info.converted_type(), ConvertedType::INT_32);
            assert_eq!(basic_info.id(), 0);
            match tp {
                Type::PrimitiveType { physical_type, .. } => {
                    assert_eq!(physical_type, PhysicalType::INT32);
                }
                Type::GroupType { .. } => panic!(),
            }
        }

        // Test illegal inputs with logical type
        result = Type::primitive_type_builder("foo", PhysicalType::INT64)
            .with_repetition(Repetition::REPEATED)
            .with_logical_type(Some(LogicalType::integer(8, true)))
            .build();
        assert!(result.is_err());
        if let Err(e) = result {
            assert_eq!(
                format!("{e}"),
                "Parquet error: Cannot annotate Integer(IntType { bit_width: 8, is_signed: true }) from INT64 for field 'foo'"
            );
        }

        // Test illegal inputs with converted type
        result = Type::primitive_type_builder("foo", PhysicalType::INT64)
            .with_repetition(Repetition::REPEATED)
            .with_converted_type(ConvertedType::BSON)
            .build();
        assert!(result.is_err());
        if let Err(e) = result {
            assert_eq!(
                format!("{e}"),
                "Parquet error: BSON cannot annotate field 'foo' because it is not a BYTE_ARRAY field"
            );
        }

        result = Type::primitive_type_builder("foo", PhysicalType::INT96)
            .with_repetition(Repetition::REQUIRED)
            .with_converted_type(ConvertedType::DECIMAL)
            .with_precision(-1)
            .with_scale(-1)
            .build();
        assert!(result.is_err());
        if let Err(e) = result {
            assert_eq!(
                format!("{e}"),
                "Parquet error: DECIMAL can only annotate INT32, INT64, BYTE_ARRAY and FIXED_LEN_BYTE_ARRAY"
            );
        }

        result = Type::primitive_type_builder("foo", PhysicalType::BYTE_ARRAY)
            .with_repetition(Repetition::REQUIRED)
            .with_logical_type(Some(LogicalType::decimal(32, 12)))
            .with_precision(-1)
            .with_scale(-1)
            .build();
        assert!(result.is_err());
        if let Err(e) = result {
            assert_eq!(
                format!("{e}"),
                "Parquet error: DECIMAL logical type scale 32 must match self.scale -1 for field 'foo'"
            );
        }

        result = Type::primitive_type_builder("foo", PhysicalType::BYTE_ARRAY)
            .with_repetition(Repetition::REQUIRED)
            .with_converted_type(ConvertedType::DECIMAL)
            .with_precision(-1)
            .with_scale(-1)
            .build();
        assert!(result.is_err());
        if let Err(e) = result {
            assert_eq!(
                format!("{e}"),
                "Parquet error: Invalid DECIMAL precision: -1"
            );
        }

        result = Type::primitive_type_builder("foo", PhysicalType::BYTE_ARRAY)
            .with_repetition(Repetition::REQUIRED)
            .with_converted_type(ConvertedType::DECIMAL)
            .with_precision(0)
            .with_scale(-1)
            .build();
        assert!(result.is_err());
        if let Err(e) = result {
            assert_eq!(
                format!("{e}"),
                "Parquet error: Invalid DECIMAL precision: 0"
            );
        }

        result = Type::primitive_type_builder("foo", PhysicalType::BYTE_ARRAY)
            .with_repetition(Repetition::REQUIRED)
            .with_converted_type(ConvertedType::DECIMAL)
            .with_precision(1)
            .with_scale(-1)
            .build();
        assert!(result.is_err());
        if let Err(e) = result {
            assert_eq!(format!("{e}"), "Parquet error: Invalid DECIMAL scale: -1");
        }

        result = Type::primitive_type_builder("foo", PhysicalType::BYTE_ARRAY)
            .with_repetition(Repetition::REQUIRED)
            .with_converted_type(ConvertedType::DECIMAL)
            .with_precision(1)
            .with_scale(2)
            .build();
        assert!(result.is_err());
        if let Err(e) = result {
            assert_eq!(
                format!("{e}"),
                "Parquet error: Invalid DECIMAL: scale (2) cannot be greater than precision (1)"
            );
        }

        // It is OK if precision == scale
        result = Type::primitive_type_builder("foo", PhysicalType::BYTE_ARRAY)
            .with_repetition(Repetition::REQUIRED)
            .with_converted_type(ConvertedType::DECIMAL)
            .with_precision(1)
            .with_scale(1)
            .build();
        assert!(result.is_ok());

        result = Type::primitive_type_builder("foo", PhysicalType::INT32)
            .with_repetition(Repetition::REQUIRED)
            .with_converted_type(ConvertedType::DECIMAL)
            .with_precision(18)
            .with_scale(2)
            .build();
        assert!(result.is_err());
        if let Err(e) = result {
            assert_eq!(
                format!("{e}"),
                "Parquet error: Cannot represent INT32 as DECIMAL with precision 18"
            );
        }

        result = Type::primitive_type_builder("foo", PhysicalType::INT64)
            .with_repetition(Repetition::REQUIRED)
            .with_converted_type(ConvertedType::DECIMAL)
            .with_precision(32)
            .with_scale(2)
            .build();
        assert!(result.is_err());
        if let Err(e) = result {
            assert_eq!(
                format!("{e}"),
                "Parquet error: Cannot represent INT64 as DECIMAL with precision 32"
            );
        }

        result = Type::primitive_type_builder("foo", PhysicalType::FIXED_LEN_BYTE_ARRAY)
            .with_repetition(Repetition::REQUIRED)
            .with_converted_type(ConvertedType::DECIMAL)
            .with_length(5)
            .with_precision(12)
            .with_scale(2)
            .build();
        assert!(result.is_err());
        if let Err(e) = result {
            assert_eq!(
                format!("{e}"),
                "Parquet error: Cannot represent FIXED_LEN_BYTE_ARRAY as DECIMAL with length 5 and precision 12. The max precision can only be 11"
            );
        }

        result = Type::primitive_type_builder("foo", PhysicalType::INT64)
            .with_repetition(Repetition::REQUIRED)
            .with_converted_type(ConvertedType::UINT_8)
            .build();
        assert!(result.is_err());
        if let Err(e) = result {
            assert_eq!(
                format!("{e}"),
                "Parquet error: UINT_8 cannot annotate field 'foo' because it is not a INT32 field"
            );
        }

        result = Type::primitive_type_builder("foo", PhysicalType::INT32)
            .with_repetition(Repetition::REQUIRED)
            .with_converted_type(ConvertedType::TIME_MICROS)
            .build();
        assert!(result.is_err());
        if let Err(e) = result {
            assert_eq!(
                format!("{e}"),
                "Parquet error: TIME_MICROS cannot annotate field 'foo' because it is not a INT64 field"
            );
        }

        result = Type::primitive_type_builder("foo", PhysicalType::BYTE_ARRAY)
            .with_repetition(Repetition::REQUIRED)
            .with_converted_type(ConvertedType::INTERVAL)
            .build();
        assert!(result.is_err());
        if let Err(e) = result {
            assert_eq!(
                format!("{e}"),
                "Parquet error: INTERVAL cannot annotate field 'foo' because it is not a FIXED_LEN_BYTE_ARRAY(12) field"
            );
        }

        result = Type::primitive_type_builder("foo", PhysicalType::FIXED_LEN_BYTE_ARRAY)
            .with_repetition(Repetition::REQUIRED)
            .with_converted_type(ConvertedType::INTERVAL)
            .with_length(1)
            .build();
        assert!(result.is_err());
        if let Err(e) = result {
            assert_eq!(
                format!("{e}"),
                "Parquet error: INTERVAL cannot annotate field 'foo' because it is not a FIXED_LEN_BYTE_ARRAY(12) field"
            );
        }

        result = Type::primitive_type_builder("foo", PhysicalType::INT32)
            .with_repetition(Repetition::REQUIRED)
            .with_converted_type(ConvertedType::ENUM)
            .build();
        assert!(result.is_err());
        if let Err(e) = result {
            assert_eq!(
                format!("{e}"),
                "Parquet error: ENUM cannot annotate field 'foo' because it is not a BYTE_ARRAY field"
            );
        }

        result = Type::primitive_type_builder("foo", PhysicalType::INT32)
            .with_repetition(Repetition::REQUIRED)
            .with_converted_type(ConvertedType::MAP)
            .build();
        assert!(result.is_err());
        if let Err(e) = result {
            assert_eq!(
                format!("{e}"),
                "Parquet error: MAP cannot be applied to primitive field 'foo'"
            );
        }

        result = Type::primitive_type_builder("foo", PhysicalType::FIXED_LEN_BYTE_ARRAY)
            .with_repetition(Repetition::REQUIRED)
            .with_converted_type(ConvertedType::DECIMAL)
            .with_length(-1)
            .build();
        assert!(result.is_err());
        if let Err(e) = result {
            assert_eq!(
                format!("{e}"),
                "Parquet error: Invalid FIXED_LEN_BYTE_ARRAY length: -1 for field 'foo'"
            );
        }

        result = Type::primitive_type_builder("foo", PhysicalType::FIXED_LEN_BYTE_ARRAY)
            .with_repetition(Repetition::REQUIRED)
            .with_logical_type(Some(LogicalType::Float16))
            .with_length(2)
            .build();
        assert!(result.is_ok());

        // Can't be other than FIXED_LEN_BYTE_ARRAY for physical type
        result = Type::primitive_type_builder("foo", PhysicalType::FLOAT)
            .with_repetition(Repetition::REQUIRED)
            .with_logical_type(Some(LogicalType::Float16))
            .with_length(2)
            .build();
        assert!(result.is_err());
        if let Err(e) = result {
            assert_eq!(
                format!("{e}"),
                "Parquet error: Cannot annotate Float16 from FLOAT for field 'foo'"
            );
        }

        // Must have length 2
        result = Type::primitive_type_builder("foo", PhysicalType::FIXED_LEN_BYTE_ARRAY)
            .with_repetition(Repetition::REQUIRED)
            .with_logical_type(Some(LogicalType::Float16))
            .with_length(4)
            .build();
        assert!(result.is_err());
        if let Err(e) = result {
            assert_eq!(
                format!("{e}"),
                "Parquet error: FLOAT16 cannot annotate field 'foo' because it is not a FIXED_LEN_BYTE_ARRAY(2) field"
            );
        }

        // Must have length 16
        result = Type::primitive_type_builder("foo", PhysicalType::FIXED_LEN_BYTE_ARRAY)
            .with_repetition(Repetition::REQUIRED)
            .with_logical_type(Some(LogicalType::Uuid))
            .with_length(15)
            .build();
        assert!(result.is_err());
        if let Err(e) = result {
            assert_eq!(
                format!("{e}"),
                "Parquet error: UUID cannot annotate field 'foo' because it is not a FIXED_LEN_BYTE_ARRAY(16) field"
            );
        }

        // test unknown logical types are ok
        result = Type::primitive_type_builder("foo", PhysicalType::BYTE_ARRAY)
            .with_logical_type(Some(LogicalType::_Unknown { field_id: 100 }))
            .build();
        assert!(result.is_ok());
    }

    #[test]
    fn test_group_type() {
        let f1 = Type::primitive_type_builder("f1", PhysicalType::INT32)
            .with_converted_type(ConvertedType::INT_32)
            .with_id(Some(0))
            .build();
        assert!(f1.is_ok());
        let f2 = Type::primitive_type_builder("f2", PhysicalType::BYTE_ARRAY)
            .with_converted_type(ConvertedType::UTF8)
            .with_id(Some(1))
            .build();
        assert!(f2.is_ok());

        let fields = vec![Arc::new(f1.unwrap()), Arc::new(f2.unwrap())];

        let result = Type::group_type_builder("foo")
            .with_repetition(Repetition::REPEATED)
            .with_logical_type(Some(LogicalType::List))
            .with_fields(fields)
            .with_id(Some(1))
            .build();
        assert!(result.is_ok());

        let tp = result.unwrap();
        let basic_info = tp.get_basic_info();
        assert!(tp.is_group());
        assert!(!tp.is_primitive());
        assert_eq!(basic_info.repetition(), Repetition::REPEATED);
        assert_eq!(basic_info.logical_type_ref(), Some(&LogicalType::List));
        assert_eq!(basic_info.converted_type(), ConvertedType::LIST);
        assert_eq!(basic_info.id(), 1);
        assert_eq!(tp.get_fields().len(), 2);
        assert_eq!(tp.get_fields()[0].name(), "f1");
        assert_eq!(tp.get_fields()[1].name(), "f2");
    }

    #[test]
    fn test_column_descriptor() {
        let result = test_column_descriptor_helper();
        assert!(
            result.is_ok(),
            "Expected result to be OK but got err:\n {}",
            result.unwrap_err()
        );
    }

    fn test_column_descriptor_helper() -> Result<()> {
        let tp = Type::primitive_type_builder("name", PhysicalType::BYTE_ARRAY)
            .with_converted_type(ConvertedType::UTF8)
            .build()?;

        let descr = ColumnDescriptor::new(Arc::new(tp), 4, 1, ColumnPath::from("name"));

        assert_eq!(descr.path(), &ColumnPath::from("name"));
        assert_eq!(descr.converted_type(), ConvertedType::UTF8);
        assert_eq!(descr.physical_type(), PhysicalType::BYTE_ARRAY);
        assert_eq!(descr.max_def_level(), 4);
        assert_eq!(descr.max_rep_level(), 1);
        assert_eq!(descr.name(), "name");
        assert_eq!(descr.type_length(), -1);
        assert_eq!(descr.type_precision(), -1);
        assert_eq!(descr.type_scale(), -1);

        Ok(())
    }

    #[test]
    fn test_schema_descriptor() {
        let result = test_schema_descriptor_helper();
        assert!(
            result.is_ok(),
            "Expected result to be OK but got err:\n {}",
            result.unwrap_err()
        );
    }

    // A helper fn to avoid handling the results from type creation
    fn test_schema_descriptor_helper() -> Result<()> {
        let mut fields = vec![];

        let inta = Type::primitive_type_builder("a", PhysicalType::INT32)
            .with_repetition(Repetition::REQUIRED)
            .with_converted_type(ConvertedType::INT_32)
            .build()?;
        fields.push(Arc::new(inta));
        let intb = Type::primitive_type_builder("b", PhysicalType::INT64)
            .with_converted_type(ConvertedType::INT_64)
            .build()?;
        fields.push(Arc::new(intb));
        let intc = Type::primitive_type_builder("c", PhysicalType::BYTE_ARRAY)
            .with_repetition(Repetition::REPEATED)
            .with_converted_type(ConvertedType::UTF8)
            .build()?;
        fields.push(Arc::new(intc));

        // 3-level list encoding
        let item1 = Type::primitive_type_builder("item1", PhysicalType::INT64)
            .with_repetition(Repetition::REQUIRED)
            .with_converted_type(ConvertedType::INT_64)
            .build()?;
        let item2 = Type::primitive_type_builder("item2", PhysicalType::BOOLEAN).build()?;
        let item3 = Type::primitive_type_builder("item3", PhysicalType::INT32)
            .with_repetition(Repetition::REPEATED)
            .with_converted_type(ConvertedType::INT_32)
            .build()?;
        let list = Type::group_type_builder("records")
            .with_repetition(Repetition::REPEATED)
            .with_converted_type(ConvertedType::LIST)
            .with_fields(vec![Arc::new(item1), Arc::new(item2), Arc::new(item3)])
            .build()?;
        let bag = Type::group_type_builder("bag")
            .with_repetition(Repetition::OPTIONAL)
            .with_fields(vec![Arc::new(list)])
            .build()?;
        fields.push(Arc::new(bag));

        let schema = Type::group_type_builder("schema")
            .with_repetition(Repetition::REPEATED)
            .with_fields(fields)
            .build()?;
        let descr = SchemaDescriptor::new(Arc::new(schema));

        let nleaves = 6;
        assert_eq!(descr.num_columns(), nleaves);

        //                             mdef mrep
        // required int32 a            0    0
        // optional int64 b            1    0
        // repeated byte_array c       1    1
        // optional group bag          1    0
        //   repeated group records    2    1
        //     required int64 item1    2    1
        //     optional boolean item2  3    1
        //     repeated int32 item3    3    2
        let ex_max_def_levels = [0, 1, 1, 2, 3, 3];
        let ex_max_rep_levels = [0, 0, 1, 1, 1, 2];

        for i in 0..nleaves {
            let col = descr.column(i);
            assert_eq!(col.max_def_level(), ex_max_def_levels[i], "{i}");
            assert_eq!(col.max_rep_level(), ex_max_rep_levels[i], "{i}");
        }

        assert_eq!(descr.column(0).path().string(), "a");
        assert_eq!(descr.column(1).path().string(), "b");
        assert_eq!(descr.column(2).path().string(), "c");
        assert_eq!(descr.column(3).path().string(), "bag.records.item1");
        assert_eq!(descr.column(4).path().string(), "bag.records.item2");
        assert_eq!(descr.column(5).path().string(), "bag.records.item3");

        assert_eq!(descr.get_column_root(0).name(), "a");
        assert_eq!(descr.get_column_root(3).name(), "bag");
        assert_eq!(descr.get_column_root(4).name(), "bag");
        assert_eq!(descr.get_column_root(5).name(), "bag");

        Ok(())
    }

    #[test]
    fn test_schema_build_tree_def_rep_levels() {
        let message_type = "
    message spark_schema {
      REQUIRED INT32 a;
      OPTIONAL group b {
        OPTIONAL INT32 _1;
        OPTIONAL INT32 _2;
      }
      OPTIONAL group c (LIST) {
        REPEATED group list {
          OPTIONAL INT32 element;
        }
      }
    }
    ";
        let schema = parse_message_type(message_type).expect("should parse schema");
        let descr = SchemaDescriptor::new(Arc::new(schema));
        // required int32 a
        assert_eq!(descr.column(0).max_def_level(), 0);
        assert_eq!(descr.column(0).max_rep_level(), 0);
        // optional int32 b._1
        assert_eq!(descr.column(1).max_def_level(), 2);
        assert_eq!(descr.column(1).max_rep_level(), 0);
        // optional int32 b._2
        assert_eq!(descr.column(2).max_def_level(), 2);
        assert_eq!(descr.column(2).max_rep_level(), 0);
        // repeated optional int32 c.list.element
        assert_eq!(descr.column(3).max_def_level(), 3);
        assert_eq!(descr.column(3).max_rep_level(), 1);
    }

    #[test]
    fn test_schema_build_tree_repeated_ancestor_def_level() {
        // Flat columns: no REPEATED ancestor → repeated_ancestor_def_level = 0
        let message_type = "
    message m {
      REQUIRED INT32 a;
      OPTIONAL INT32 b;
      OPTIONAL group s {
        OPTIONAL INT32 x;
      }
    }
    ";
        let schema = parse_message_type(message_type).expect("should parse schema");
        let descr = SchemaDescriptor::new(Arc::new(schema));
        assert_eq!(descr.column(0).repeated_ancestor_def_level(), 0); // a
        assert_eq!(descr.column(1).repeated_ancestor_def_level(), 0); // b
        assert_eq!(descr.column(2).repeated_ancestor_def_level(), 0); // s.x

        // Standard list: OPTIONAL outer, REPEATED group, OPTIONAL element
        // repeated_ancestor_def_level is the def_level at the REPEATED group (= 2)
        let message_type = "
    message m {
      OPTIONAL group c (LIST) {
        REPEATED group list {
          OPTIONAL INT32 element;
        }
      }
    }
    ";
        let schema = parse_message_type(message_type).expect("should parse schema");
        let descr = SchemaDescriptor::new(Arc::new(schema));
        // c(optional)=1, list(repeated)=2, element(optional)=3
        assert_eq!(descr.column(0).max_def_level(), 3);
        assert_eq!(descr.column(0).max_rep_level(), 1);
        assert_eq!(descr.column(0).repeated_ancestor_def_level(), 2);

        // Required list: REQUIRED outer, REPEATED group, REQUIRED element
        // No OPTIONAL nodes between REPEATED and leaf, so repeated_ancestor_def_level == max_def_level
        let message_type = "
    message m {
      REQUIRED group c (LIST) {
        REPEATED group list {
          REQUIRED INT32 element;
        }
      }
    }
    ";
        let schema = parse_message_type(message_type).expect("should parse schema");
        let descr = SchemaDescriptor::new(Arc::new(schema));
        // list(repeated)=1, element(required)=1
        assert_eq!(descr.column(0).max_def_level(), 1);
        assert_eq!(descr.column(0).max_rep_level(), 1);
        assert_eq!(descr.column(0).repeated_ancestor_def_level(), 1);

        // Nested lists: innermost REPEATED wins
        let message_type = "
    message m {
      OPTIONAL group outer (LIST) {
        REPEATED group list {
          OPTIONAL group inner (LIST) {
            REPEATED group list2 {
              OPTIONAL INT32 element;
            }
          }
        }
      }
    }
    ";
        let schema = parse_message_type(message_type).expect("should parse schema");
        let descr = SchemaDescriptor::new(Arc::new(schema));
        // outer(opt)=1, list(rep)=2, inner(opt)=3, list2(rep)=4, element(opt)=5
        assert_eq!(descr.column(0).max_def_level(), 5);
        assert_eq!(descr.column(0).max_rep_level(), 2);
        assert_eq!(descr.column(0).repeated_ancestor_def_level(), 4);

        // Struct inside list: all sibling leaves share the same repeated_ancestor_def_level
        let message_type = "
    message m {
      OPTIONAL group bag (LIST) {
        REPEATED group list {
          REQUIRED group item {
            OPTIONAL INT32 x;
            REQUIRED INT32 y;
          }
        }
      }
    }
    ";
        let schema = parse_message_type(message_type).expect("should parse schema");
        let descr = SchemaDescriptor::new(Arc::new(schema));
        // bag(opt)=1, list(rep)=2, item(req)=2, x(opt)=3
        assert_eq!(descr.column(0).repeated_ancestor_def_level(), 2); // bag.list.item.x
        // bag(opt)=1, list(rep)=2, item(req)=2, y(req)=2
        assert_eq!(descr.column(1).repeated_ancestor_def_level(), 2); // bag.list.item.y

        // Map type: key (required) and value (optional) under the same REPEATED group
        let message_type = "
    message m {
      OPTIONAL group my_map (MAP) {
        REPEATED group key_value {
          REQUIRED BYTE_ARRAY key (UTF8);
          OPTIONAL INT32 value;
        }
      }
    }
    ";
        let schema = parse_message_type(message_type).expect("should parse schema");
        let descr = SchemaDescriptor::new(Arc::new(schema));
        // my_map(opt)=1, key_value(rep)=2, key(req)=2
        assert_eq!(descr.column(0).max_def_level(), 2);
        assert_eq!(descr.column(0).repeated_ancestor_def_level(), 2); // key: max_def == repeated_ancestor
        // my_map(opt)=1, key_value(rep)=2, value(opt)=3
        assert_eq!(descr.column(1).max_def_level(), 3);
        assert_eq!(descr.column(1).repeated_ancestor_def_level(), 2); // value: max_def > repeated_ancestor
    }

    #[test]
    #[should_panic(expected = "Cannot call get_physical_type() on a non-primitive type")]
    fn test_get_physical_type_panic() {
        let list = Type::group_type_builder("records")
            .with_repetition(Repetition::REPEATED)
            .build()
            .unwrap();
        list.get_physical_type();
    }

    #[test]
    fn test_get_physical_type_primitive() {
        let f = Type::primitive_type_builder("f", PhysicalType::INT64)
            .build()
            .unwrap();
        assert_eq!(f.get_physical_type(), PhysicalType::INT64);

        let f = Type::primitive_type_builder("f", PhysicalType::BYTE_ARRAY)
            .build()
            .unwrap();
        assert_eq!(f.get_physical_type(), PhysicalType::BYTE_ARRAY);
    }

    #[test]
    fn test_check_contains_primitive_primitive() {
        // OK
        let f1 = Type::primitive_type_builder("f", PhysicalType::INT32)
            .build()
            .unwrap();
        let f2 = Type::primitive_type_builder("f", PhysicalType::INT32)
            .build()
            .unwrap();
        assert!(f1.check_contains(&f2));

        // OK: different logical type does not affect check_contains
        let f1 = Type::primitive_type_builder("f", PhysicalType::INT32)
            .with_converted_type(ConvertedType::UINT_8)
            .build()
            .unwrap();
        let f2 = Type::primitive_type_builder("f", PhysicalType::INT32)
            .with_converted_type(ConvertedType::UINT_16)
            .build()
            .unwrap();
        assert!(f1.check_contains(&f2));

        // KO: different name
        let f1 = Type::primitive_type_builder("f1", PhysicalType::INT32)
            .build()
            .unwrap();
        let f2 = Type::primitive_type_builder("f2", PhysicalType::INT32)
            .build()
            .unwrap();
        assert!(!f1.check_contains(&f2));

        // KO: different type
        let f1 = Type::primitive_type_builder("f", PhysicalType::INT32)
            .build()
            .unwrap();
        let f2 = Type::primitive_type_builder("f", PhysicalType::INT64)
            .build()
            .unwrap();
        assert!(!f1.check_contains(&f2));

        // KO: different repetition
        let f1 = Type::primitive_type_builder("f", PhysicalType::INT32)
            .with_repetition(Repetition::REQUIRED)
            .build()
            .unwrap();
        let f2 = Type::primitive_type_builder("f", PhysicalType::INT32)
            .with_repetition(Repetition::OPTIONAL)
            .build()
            .unwrap();
        assert!(!f1.check_contains(&f2));
    }

    // function to create a new group type for testing
    fn test_new_group_type(name: &str, repetition: Repetition, types: Vec<Type>) -> Type {
        Type::group_type_builder(name)
            .with_repetition(repetition)
            .with_fields(types.into_iter().map(Arc::new).collect())
            .build()
            .unwrap()
    }

    #[test]
    fn test_check_contains_group_group() {
        // OK: should match okay with empty fields
        let f1 = Type::group_type_builder("f").build().unwrap();
        let f2 = Type::group_type_builder("f").build().unwrap();
        assert!(f1.check_contains(&f2));
        assert!(!f1.is_optional());

        // OK: fields match
        let f1 = test_new_group_type(
            "f",
            Repetition::REPEATED,
            vec![
                Type::primitive_type_builder("f1", PhysicalType::INT32)
                    .build()
                    .unwrap(),
                Type::primitive_type_builder("f2", PhysicalType::INT64)
                    .build()
                    .unwrap(),
            ],
        );
        let f2 = test_new_group_type(
            "f",
            Repetition::REPEATED,
            vec![
                Type::primitive_type_builder("f1", PhysicalType::INT32)
                    .build()
                    .unwrap(),
                Type::primitive_type_builder("f2", PhysicalType::INT64)
                    .build()
                    .unwrap(),
            ],
        );
        assert!(f1.check_contains(&f2));

        // OK: subset of fields
        let f1 = test_new_group_type(
            "f",
            Repetition::REPEATED,
            vec![
                Type::primitive_type_builder("f1", PhysicalType::INT32)
                    .build()
                    .unwrap(),
                Type::primitive_type_builder("f2", PhysicalType::INT64)
                    .build()
                    .unwrap(),
            ],
        );
        let f2 = test_new_group_type(
            "f",
            Repetition::REPEATED,
            vec![
                Type::primitive_type_builder("f2", PhysicalType::INT64)
                    .build()
                    .unwrap(),
            ],
        );
        assert!(f1.check_contains(&f2));

        // KO: different name
        let f1 = Type::group_type_builder("f1").build().unwrap();
        let f2 = Type::group_type_builder("f2").build().unwrap();
        assert!(!f1.check_contains(&f2));

        // KO: different repetition
        let f1 = Type::group_type_builder("f")
            .with_repetition(Repetition::OPTIONAL)
            .build()
            .unwrap();
        let f2 = Type::group_type_builder("f")
            .with_repetition(Repetition::REPEATED)
            .build()
            .unwrap();
        assert!(!f1.check_contains(&f2));

        // KO: different fields
        let f1 = test_new_group_type(
            "f",
            Repetition::REPEATED,
            vec![
                Type::primitive_type_builder("f1", PhysicalType::INT32)
                    .build()
                    .unwrap(),
                Type::primitive_type_builder("f2", PhysicalType::INT64)
                    .build()
                    .unwrap(),
            ],
        );
        let f2 = test_new_group_type(
            "f",
            Repetition::REPEATED,
            vec![
                Type::primitive_type_builder("f1", PhysicalType::INT32)
                    .build()
                    .unwrap(),
                Type::primitive_type_builder("f2", PhysicalType::BOOLEAN)
                    .build()
                    .unwrap(),
            ],
        );
        assert!(!f1.check_contains(&f2));

        // KO: different fields
        let f1 = test_new_group_type(
            "f",
            Repetition::REPEATED,
            vec![
                Type::primitive_type_builder("f1", PhysicalType::INT32)
                    .build()
                    .unwrap(),
                Type::primitive_type_builder("f2", PhysicalType::INT64)
                    .build()
                    .unwrap(),
            ],
        );
        let f2 = test_new_group_type(
            "f",
            Repetition::REPEATED,
            vec![
                Type::primitive_type_builder("f3", PhysicalType::INT32)
                    .build()
                    .unwrap(),
            ],
        );
        assert!(!f1.check_contains(&f2));
    }

    #[test]
    fn test_check_contains_group_primitive() {
        // KO: should not match
        let f1 = Type::group_type_builder("f").build().unwrap();
        let f2 = Type::primitive_type_builder("f", PhysicalType::INT64)
            .build()
            .unwrap();
        assert!(!f1.check_contains(&f2));
        assert!(!f2.check_contains(&f1));

        // KO: should not match when primitive field is part of group type
        let f1 = test_new_group_type(
            "f",
            Repetition::REPEATED,
            vec![
                Type::primitive_type_builder("f1", PhysicalType::INT32)
                    .build()
                    .unwrap(),
            ],
        );
        let f2 = Type::primitive_type_builder("f1", PhysicalType::INT32)
            .build()
            .unwrap();
        assert!(!f1.check_contains(&f2));
        assert!(!f2.check_contains(&f1));

        // OK: match nested types
        let f1 = test_new_group_type(
            "a",
            Repetition::REPEATED,
            vec![
                test_new_group_type(
                    "b",
                    Repetition::REPEATED,
                    vec![
                        Type::primitive_type_builder("c", PhysicalType::INT32)
                            .build()
                            .unwrap(),
                    ],
                ),
                Type::primitive_type_builder("d", PhysicalType::INT64)
                    .build()
                    .unwrap(),
                Type::primitive_type_builder("e", PhysicalType::BOOLEAN)
                    .build()
                    .unwrap(),
            ],
        );
        let f2 = test_new_group_type(
            "a",
            Repetition::REPEATED,
            vec![test_new_group_type(
                "b",
                Repetition::REPEATED,
                vec![
                    Type::primitive_type_builder("c", PhysicalType::INT32)
                        .build()
                        .unwrap(),
                ],
            )],
        );
        assert!(f1.check_contains(&f2)); // should match
        assert!(!f2.check_contains(&f1)); // should fail
    }

    #[test]
    fn test_schema_type_thrift_conversion_err() {
        let schema = Type::primitive_type_builder("col", PhysicalType::INT32)
            .build()
            .unwrap();
        let schema = Arc::new(schema);
        let thrift_schema = schema_to_buf(&schema);
        assert!(thrift_schema.is_err());
        if let Err(e) = thrift_schema {
            assert_eq!(
                format!("{e}"),
                "Parquet error: Root schema must be Group type"
            );
        }
    }

    #[test]
    fn test_schema_type_thrift_conversion() {
        let message_type = "
    message conversions {
      REQUIRED INT64 id;
      OPTIONAL FIXED_LEN_BYTE_ARRAY (2) f16 (FLOAT16);
      OPTIONAL group int_array_Array (LIST) {
        REPEATED group list {
          OPTIONAL group element (LIST) {
            REPEATED group list {
              OPTIONAL INT32 element;
            }
          }
        }
      }
      OPTIONAL group int_map (MAP) {
        REPEATED group map (MAP_KEY_VALUE) {
          REQUIRED BYTE_ARRAY key (UTF8);
          OPTIONAL INT32 value;
        }
      }
      OPTIONAL group int_Map_Array (LIST) {
        REPEATED group list {
          OPTIONAL group g (MAP) {
            REPEATED group map (MAP_KEY_VALUE) {
              REQUIRED BYTE_ARRAY key (UTF8);
              OPTIONAL group value {
                OPTIONAL group H {
                  OPTIONAL group i (LIST) {
                    REPEATED group list {
                      OPTIONAL DOUBLE element;
                    }
                  }
                }
              }
            }
          }
        }
      }
      OPTIONAL group nested_struct {
        OPTIONAL INT32 A;
        OPTIONAL group b (LIST) {
          REPEATED group list {
            REQUIRED FIXED_LEN_BYTE_ARRAY (16) element;
          }
        }
      }
    }
    ";
        let expected_schema = parse_message_type(message_type).unwrap();
        let result_schema = roundtrip_schema(Arc::new(expected_schema.clone())).unwrap();
        assert_eq!(result_schema, Arc::new(expected_schema));
    }

    #[test]
    fn test_schema_type_thrift_conversion_decimal() {
        let message_type = "
    message decimals {
      OPTIONAL INT32 field0;
      OPTIONAL INT64 field1 (DECIMAL (18, 2));
      OPTIONAL FIXED_LEN_BYTE_ARRAY (16) field2 (DECIMAL (38, 18));
      OPTIONAL BYTE_ARRAY field3 (DECIMAL (9));
    }
    ";
        let expected_schema = parse_message_type(message_type).unwrap();
        let result_schema = roundtrip_schema(Arc::new(expected_schema.clone())).unwrap();
        assert_eq!(result_schema, Arc::new(expected_schema));
    }

    // Tests schema conversion from thrift, when num_children is set to Some(0) for a
    // primitive type.
    #[test]
    fn test_schema_from_thrift_with_num_children_set() {
        // schema definition written by parquet-cpp version 1.3.2-SNAPSHOT
        let message_type = "
    message schema {
      OPTIONAL BYTE_ARRAY id (UTF8);
      OPTIONAL BYTE_ARRAY name (UTF8);
      OPTIONAL BYTE_ARRAY message (UTF8);
      OPTIONAL INT32 type (UINT_8);
      OPTIONAL INT64 author_time (TIMESTAMP_MILLIS);
      OPTIONAL INT64 __index_level_0__;
    }
    ";

        let expected_schema = Arc::new(parse_message_type(message_type).unwrap());
        let mut buf = schema_to_buf(&expected_schema).unwrap();
        let mut thrift_schema = buf_to_schema_list(&mut buf).unwrap();

        // Change all of None to Some(0)
        for elem in &mut thrift_schema[..] {
            if elem.num_children.is_none() {
                elem.num_children = Some(0);
            }
        }

        let result_schema = parquet_schema_from_array(thrift_schema).unwrap();
        assert_eq!(result_schema, expected_schema);
    }

    // Sometimes parquet-cpp sets repetition level for the root node, which is against
    // the format definition, but we need to handle it by setting it back to None.
    #[test]
    fn test_schema_from_thrift_root_has_repetition() {
        // schema definition written by parquet-cpp version 1.3.2-SNAPSHOT
        let message_type = "
    message schema {
      OPTIONAL BYTE_ARRAY a (UTF8);
      OPTIONAL INT32 b (UINT_8);
    }
    ";

        let expected_schema = Arc::new(parse_message_type(message_type).unwrap());
        let mut buf = schema_to_buf(&expected_schema).unwrap();
        let mut thrift_schema = buf_to_schema_list(&mut buf).unwrap();
        thrift_schema[0].repetition_type = Some(Repetition::REQUIRED);

        let result_schema = parquet_schema_from_array(thrift_schema).unwrap();
        assert_eq!(result_schema, expected_schema);
    }

    #[test]
    fn test_schema_from_thrift_group_has_no_child() {
        let message_type = "message schema {}";

        let expected_schema = Arc::new(parse_message_type(message_type).unwrap());
        let mut buf = schema_to_buf(&expected_schema).unwrap();
        let mut thrift_schema = buf_to_schema_list(&mut buf).unwrap();
        thrift_schema[0].repetition_type = Some(Repetition::REQUIRED);

        let result_schema = parquet_schema_from_array(thrift_schema).unwrap();
        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 {
            r#type: None,
            type_length: None,
            repetition_type: Some(Repetition::REQUIRED),
            name: "schema",
            num_children: Some(-1),
            converted_type: None,
            scale: None,
            precision: None,
            field_id: None,
            logical_type: None,
        }];
        let result = parquet_schema_from_array(elements);
        assert!(result.unwrap_err().to_string().contains("Integer overflow"));
    }
}