parquet/arrow/schema/

mod.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.

//! Converting Parquet schema <--> Arrow schema: [`ArrowSchemaConverter`] and [parquet_to_arrow_schema]

use base64::Engine;
use base64::prelude::BASE64_STANDARD;
use std::collections::HashMap;
use std::sync::Arc;

use arrow_ipc::writer;
use arrow_schema::{DataType, Field, FieldRef, Fields, Schema, TimeUnit};

use crate::basic::{
    ConvertedType, LogicalType, Repetition, TimeUnit as ParquetTimeUnit, Type as PhysicalType,
};
use crate::errors::{ParquetError, Result};
use crate::file::{metadata::KeyValue, properties::WriterProperties};
use crate::schema::types::{ColumnDescriptor, SchemaDescriptor, Type};

mod complex;
mod extension;
mod primitive;
pub mod virtual_type;

use super::PARQUET_FIELD_ID_META_KEY;
use crate::arrow::ProjectionMask;
use crate::arrow::schema::extension::{
    has_extension_type, logical_type_for_binary, logical_type_for_binary_view,
    logical_type_for_fixed_size_binary, logical_type_for_string, logical_type_for_struct,
    try_add_extension_type,
};
pub(crate) use complex::{ParquetField, ParquetFieldType, VirtualColumnType};

/// Convert Parquet schema to Arrow schema including optional metadata
///
/// Attempts to decode any existing Arrow schema metadata, falling back
/// to converting the Parquet schema column-wise
pub fn parquet_to_arrow_schema(
    parquet_schema: &SchemaDescriptor,
    key_value_metadata: Option<&Vec<KeyValue>>,
) -> Result<Schema> {
    parquet_to_arrow_schema_by_columns(parquet_schema, ProjectionMask::all(), key_value_metadata)
}

/// Convert parquet schema to arrow schema including optional metadata,
/// only preserving some leaf columns.
pub fn parquet_to_arrow_schema_by_columns(
    parquet_schema: &SchemaDescriptor,
    mask: ProjectionMask,
    key_value_metadata: Option<&Vec<KeyValue>>,
) -> Result<Schema> {
    Ok(parquet_to_arrow_schema_and_fields(parquet_schema, mask, key_value_metadata, &[])?.0)
}

/// Determines the Arrow Schema from a Parquet schema
///
/// Looks for an Arrow schema metadata "hint" (see
/// [`parquet_to_arrow_field_levels`]), and uses it if present to ensure
/// lossless round trips.
pub(crate) fn parquet_to_arrow_schema_and_fields(
    parquet_schema: &SchemaDescriptor,
    mask: ProjectionMask,
    key_value_metadata: Option<&Vec<KeyValue>>,
    virtual_columns: &[FieldRef],
) -> Result<(Schema, Option<ParquetField>)> {
    let mut metadata = parse_key_value_metadata(key_value_metadata).unwrap_or_default();
    let maybe_schema = metadata
        .remove(super::ARROW_SCHEMA_META_KEY)
        .map(|value| get_arrow_schema_from_metadata(&value))
        .transpose()?;

    // Add the Arrow metadata to the Parquet metadata skipping keys that collide
    if let Some(arrow_schema) = &maybe_schema {
        arrow_schema.metadata().iter().for_each(|(k, v)| {
            metadata.entry(k.clone()).or_insert_with(|| v.clone());
        });
    }

    let hint = maybe_schema.as_ref().map(|s| s.fields());
    let field_levels =
        parquet_to_arrow_field_levels_with_virtual(parquet_schema, mask, hint, virtual_columns)?;
    let schema = Schema::new_with_metadata(field_levels.fields, metadata);
    Ok((schema, field_levels.levels))
}

/// Schema information necessary to decode a parquet file as arrow [`Fields`]
///
/// In particular this stores the dremel-level information necessary to correctly
/// interpret the encoded definition and repetition levels
///
/// Note: this is an opaque container intended to be used with lower-level APIs
/// within this crate
#[derive(Debug, Clone)]
pub struct FieldLevels {
    pub(crate) fields: Fields,
    pub(crate) levels: Option<ParquetField>,
}

/// Convert a parquet [`SchemaDescriptor`] to [`FieldLevels`]
///
/// Columns not included within [`ProjectionMask`] will be ignored.
///
/// The optional `hint` parameter is the desired Arrow schema. See the
/// [`arrow`] module documentation for more information.
///
/// [`arrow`]: crate::arrow
///
/// # Notes:
/// Where a field type in `hint` is compatible with the corresponding parquet type in `schema`, it
/// will be used, otherwise the default arrow type for the given parquet column type will be used.
///
/// This is to accommodate arrow types that cannot be round-tripped through parquet natively.
/// Depending on the parquet writer, this can lead to a mismatch between a file's parquet schema
/// and its embedded arrow schema. The parquet `schema` must be treated as authoritative in such
/// an event. See [#1663](https://github.com/apache/arrow-rs/issues/1663) for more information
///
/// Note: this is a low-level API, most users will want to make use of the higher-level
/// [`parquet_to_arrow_schema`] for decoding metadata from a parquet file.
pub fn parquet_to_arrow_field_levels(
    schema: &SchemaDescriptor,
    mask: ProjectionMask,
    hint: Option<&Fields>,
) -> Result<FieldLevels> {
    parquet_to_arrow_field_levels_with_virtual(schema, mask, hint, &[])
}

/// Convert a parquet [`SchemaDescriptor`] to [`FieldLevels`] with support for virtual columns
///
/// Columns not included within [`ProjectionMask`] will be ignored.
///
/// The optional `hint` parameter is the desired Arrow schema. See the
/// [`arrow`] module documentation for more information.
///
/// [`arrow`]: crate::arrow
///
/// # Arguments
/// * `schema` - The Parquet schema descriptor
/// * `mask` - Projection mask to select which columns to include
/// * `hint` - Optional hint for Arrow field types to use instead of defaults
/// * `virtual_columns` - Virtual columns to append to the schema (e.g., row numbers)
///
/// # Notes:
/// Where a field type in `hint` is compatible with the corresponding parquet type in `schema`, it
/// will be used, otherwise the default arrow type for the given parquet column type will be used.
///
/// Virtual columns are columns that don't exist in the Parquet file but are generated during reading.
/// They must have extension type names starting with "arrow.virtual.".
///
/// This is to accommodate arrow types that cannot be round-tripped through parquet natively.
/// Depending on the parquet writer, this can lead to a mismatch between a file's parquet schema
/// and its embedded arrow schema. The parquet `schema` must be treated as authoritative in such
/// an event. See [#1663](https://github.com/apache/arrow-rs/issues/1663) for more information
///
/// Note: this is a low-level API, most users will want to make use of the higher-level
/// [`parquet_to_arrow_schema`] for decoding metadata from a parquet file.
pub fn parquet_to_arrow_field_levels_with_virtual(
    schema: &SchemaDescriptor,
    mask: ProjectionMask,
    hint: Option<&Fields>,
    virtual_columns: &[FieldRef],
) -> Result<FieldLevels> {
    // Validate that all fields are virtual columns
    for field in virtual_columns {
        if !virtual_type::is_virtual_column(field) {
            return Err(ParquetError::General(format!(
                "Field '{}' is not a virtual column. Virtual columns must have extension type names starting with 'arrow.virtual.'",
                field.name()
            )));
        }
    }

    // Convert the regular schema first
    let mut parquet_field = match complex::convert_schema(schema, mask, hint)? {
        Some(field) => field,
        None if virtual_columns.is_empty() => {
            return Ok(FieldLevels {
                fields: Fields::empty(),
                levels: None,
            });
        }
        None => {
            // No regular fields, but we have virtual columns - create empty root struct
            ParquetField {
                rep_level: 0,
                def_level: 0,
                nullable: false,
                arrow_type: DataType::Struct(Fields::empty()),
                field_type: ParquetFieldType::Group {
                    children: Vec::new(),
                },
            }
        }
    };

    // Append virtual columns if any
    if !virtual_columns.is_empty() {
        match &mut parquet_field.field_type {
            ParquetFieldType::Group { children } => {
                // Get the mutable fields from the struct type
                let DataType::Struct(ref mut fields) = parquet_field.arrow_type else {
                    unreachable!("Root field must be a struct");
                };

                // Convert to mutable Vec to append
                let mut fields_vec: Vec<FieldRef> = fields.iter().cloned().collect();

                // Append each virtual column
                for virtual_column in virtual_columns {
                    // Virtual columns can only be added at the root level
                    assert_eq!(
                        parquet_field.rep_level, 0,
                        "Virtual columns can only be added at rep level 0"
                    );
                    assert_eq!(
                        parquet_field.def_level, 0,
                        "Virtual columns can only be added at def level 0"
                    );

                    fields_vec.push(virtual_column.clone());
                    let virtual_parquet_field = complex::convert_virtual_field(
                        virtual_column,
                        parquet_field.rep_level,
                        parquet_field.def_level,
                    )?;
                    children.push(virtual_parquet_field);
                }

                // Update the fields
                parquet_field.arrow_type = DataType::Struct(Fields::from(fields_vec));
            }
            _ => unreachable!("Root field must be a group"),
        }
    }

    match &parquet_field.arrow_type {
        DataType::Struct(fields) => Ok(FieldLevels {
            fields: fields.clone(),
            levels: Some(parquet_field),
        }),
        _ => unreachable!(),
    }
}

/// Try to convert Arrow schema metadata into a schema
fn get_arrow_schema_from_metadata(encoded_meta: &str) -> Result<Schema> {
    let decoded = BASE64_STANDARD.decode(encoded_meta);
    match decoded {
        Ok(bytes) => {
            let slice = if bytes.len() > 8 && bytes[0..4] == [255u8; 4] {
                &bytes[8..]
            } else {
                bytes.as_slice()
            };
            match arrow_ipc::root_as_message(slice) {
                Ok(message) => {
                    let schema = message
                        .header_as_schema()
                        .ok_or_else(|| arrow_err!("the message is not Arrow Schema"))?;
                    arrow_ipc::convert::try_fb_to_schema(schema).map_err(Into::into)
                }
                Err(err) => {
                    // The flatbuffers implementation returns an error on verification error.
                    Err(arrow_err!(
                        "Unable to get root as message stored in {}: {:?}",
                        super::ARROW_SCHEMA_META_KEY,
                        err
                    ))
                }
            }
        }
        Err(err) => {
            // The C++ implementation returns an error if the schema can't be parsed.
            Err(arrow_err!(
                "Unable to decode the encoded schema stored in {}, {:?}",
                super::ARROW_SCHEMA_META_KEY,
                err
            ))
        }
    }
}

/// Encodes the Arrow schema into the IPC format, and base64 encodes it
pub fn encode_arrow_schema(schema: &Schema) -> String {
    let options = writer::IpcWriteOptions::default();
    let mut dictionary_tracker = writer::DictionaryTracker::new(true);
    let data_gen = writer::IpcDataGenerator::default();
    let mut serialized_schema =
        data_gen.schema_to_bytes_with_dictionary_tracker(schema, &mut dictionary_tracker, &options);

    // manually prepending the length to the schema as arrow uses the legacy IPC format
    // TODO: change after addressing ARROW-9777
    let schema_len = serialized_schema.ipc_message.len();
    let mut len_prefix_schema = Vec::with_capacity(schema_len + 8);
    len_prefix_schema.append(&mut vec![255u8, 255, 255, 255]);
    len_prefix_schema.append((schema_len as u32).to_le_bytes().to_vec().as_mut());
    len_prefix_schema.append(&mut serialized_schema.ipc_message);

    BASE64_STANDARD.encode(&len_prefix_schema)
}

/// Reject map keys that are nullable after resolving their encoded layout.
/// This validates the writer's logical contract, not a batch's physical layout.
pub(crate) fn validate_map_key_type(data_type: &DataType) -> Result<()> {
    match data_type {
        DataType::Map(entries, _) => {
            let DataType::Struct(fields) = entries.data_type() else {
                return Err(arrow_err!(
                    "DataType::Map should contain a struct field child"
                ));
            };
            if fields.len() != 2 {
                return Err(arrow_err!("DataType::Map entries must contain two fields"));
            }
            let key = &fields[0];
            let mut key_type = key.data_type();
            let mut nullable = key.is_nullable();
            loop {
                match key_type {
                    DataType::Dictionary(_, value) => key_type = value,
                    DataType::RunEndEncoded(_, value) => {
                        nullable |= value.is_nullable();
                        key_type = value.data_type();
                    }
                    _ => break,
                }
            }
            if nullable {
                return Err(arrow_err!(
                    "Map key field '{}' must be non-nullable, including its encoded values",
                    key.name()
                ));
            }
            validate_map_key_type(entries.data_type())?;
        }
        DataType::Struct(fields) => {
            for field in fields {
                validate_map_key_type(field.data_type())?;
            }
        }
        DataType::List(field)
        | DataType::LargeList(field)
        | DataType::FixedSizeList(field, _)
        | DataType::ListView(field)
        | DataType::LargeListView(field)
        | DataType::RunEndEncoded(_, field) => validate_map_key_type(field.data_type())?,
        DataType::Dictionary(_, value) => validate_map_key_type(value)?,
        _ => {}
    }
    Ok(())
}

/// Normalize `field` for Arrow schema metadata.
///
/// This removes wrappers the Parquet reader does not reconstruct: REE at any
/// nesting depth and dictionaries around nested values. Scalar dictionaries,
/// including `Utf8View` and `BinaryView`, are retained in the metadata.
fn normalize_metadata_field(field: &Field) -> Field {
    match field.data_type() {
        DataType::RunEndEncoded(_, value_field) => {
            let value = normalize_metadata_field(value_field);
            field
                .clone()
                .with_data_type(value.data_type().clone())
                .with_nullable(field.is_nullable() || value.is_nullable())
        }
        DataType::Dictionary(key, value_type) => {
            // Dictionary values have no Field, so use a synthetic one to retain
            // the nullability of any REE value while normalizing recursively.
            let value_field = Field::new("", value_type.as_ref().clone(), false);
            let value = normalize_metadata_field(&value_field);
            let value_type = value.data_type();
            // Parquet expands nested dictionary values into leaf columns.
            // Scalar view dictionaries are supported by the reader and must remain.
            let data_type = if value_type.is_nested() {
                value_type.clone()
            } else {
                DataType::Dictionary(key.clone(), Box::new(value_type.clone()))
            };
            field
                .clone()
                .with_data_type(data_type)
                .with_nullable(field.is_nullable() || value.is_nullable())
        }
        _ => field
            .clone()
            .with_data_type(normalize_metadata_children(field.data_type())),
    }
}

fn normalize_metadata_children(dt: &DataType) -> DataType {
    match dt {
        DataType::Struct(fields) => DataType::Struct(
            fields
                .iter()
                .map(|f| Arc::new(normalize_metadata_field(f)))
                .collect(),
        ),
        DataType::List(f) => DataType::List(Arc::new(normalize_metadata_field(f))),
        DataType::LargeList(f) => DataType::LargeList(Arc::new(normalize_metadata_field(f))),
        DataType::FixedSizeList(f, n) => {
            DataType::FixedSizeList(Arc::new(normalize_metadata_field(f)), *n)
        }
        DataType::ListView(f) => DataType::ListView(Arc::new(normalize_metadata_field(f))),
        DataType::LargeListView(f) => {
            DataType::LargeListView(Arc::new(normalize_metadata_field(f)))
        }
        DataType::Map(f, sorted) => DataType::Map(Arc::new(normalize_metadata_field(f)), *sorted),
        _ => dt.clone(),
    }
}

/// Whether `dt` needs normalization before being written as Arrow schema metadata.
fn needs_metadata_normalization(dt: &DataType) -> bool {
    match dt {
        DataType::RunEndEncoded(_, _) => true,
        DataType::Struct(fields) => fields
            .iter()
            .any(|f| needs_metadata_normalization(f.data_type())),
        DataType::List(f)
        | DataType::LargeList(f)
        | DataType::FixedSizeList(f, _)
        | DataType::ListView(f)
        | DataType::LargeListView(f)
        | DataType::Map(f, _) => needs_metadata_normalization(f.data_type()),
        DataType::Dictionary(_, value) => value.is_nested() || needs_metadata_normalization(value),
        _ => false,
    }
}

/// Mutates writer metadata by storing the encoded Arrow schema hint in
/// [`ARROW_SCHEMA_META_KEY`].
///
/// If there is an existing Arrow schema metadata, it is replaced.
///
/// [`ARROW_SCHEMA_META_KEY`]: crate::arrow::ARROW_SCHEMA_META_KEY
pub fn add_encoded_arrow_schema_to_metadata(schema: &Schema, props: &mut WriterProperties) {
    let needs_normalization = schema
        .fields()
        .iter()
        .any(|f| needs_metadata_normalization(f.data_type()));
    let metadata_schema;
    let schema = if needs_normalization {
        let fields: Vec<Field> = schema
            .fields()
            .iter()
            .map(|f| normalize_metadata_field(f))
            .collect();
        metadata_schema = Schema::new_with_metadata(fields, schema.metadata().clone());
        &metadata_schema
    } else {
        schema
    };
    let encoded = encode_arrow_schema(schema);

    let schema_kv = KeyValue {
        key: super::ARROW_SCHEMA_META_KEY.to_string(),
        value: Some(encoded),
    };

    let meta = props
        .key_value_metadata
        .get_or_insert_with(Default::default);

    // check if ARROW:schema exists, and overwrite it
    let schema_meta = meta
        .iter()
        .enumerate()
        .find(|(_, kv)| kv.key.as_str() == super::ARROW_SCHEMA_META_KEY);
    match schema_meta {
        Some((i, _)) => {
            meta.remove(i);
            meta.push(schema_kv);
        }
        None => {
            meta.push(schema_kv);
        }
    }
}

/// Converter for Arrow schema to Parquet schema
///
/// See the documentation on the [`arrow`] module for background
/// information on how Arrow schema is represented in Parquet.
///
/// [`arrow`]: crate::arrow
///
/// # Example:
/// ```
/// # use std::sync::Arc;
/// # use arrow_schema::{Field, Schema, DataType};
/// # use parquet::arrow::ArrowSchemaConverter;
/// use parquet::schema::types::{SchemaDescriptor, Type};
/// use parquet::basic; // note there are two `Type`s in the following example
/// // create an Arrow Schema
/// let arrow_schema = Schema::new(vec![
///   Field::new("a", DataType::Int64, true),
///   Field::new("b", DataType::Date32, true),
/// ]);
/// // convert the Arrow schema to a Parquet schema
/// let parquet_schema = ArrowSchemaConverter::new()
///   .convert(&arrow_schema)
///   .unwrap();
///
/// let expected_parquet_schema = SchemaDescriptor::new(
///   Arc::new(
///     Type::group_type_builder("arrow_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()
///   )
/// );
/// assert_eq!(parquet_schema, expected_parquet_schema);
/// ```
#[derive(Debug)]
pub struct ArrowSchemaConverter<'a> {
    /// Name of the root schema in Parquet
    schema_root: &'a str,
    /// Should we coerce Arrow types to compatible Parquet types?
    ///
    /// See docs on [`Self::with_coerce_types`]
    coerce_types: bool,
}

impl Default for ArrowSchemaConverter<'_> {
    fn default() -> Self {
        Self::new()
    }
}

impl<'a> ArrowSchemaConverter<'a> {
    /// Create a new converter
    pub fn new() -> Self {
        Self {
            schema_root: "arrow_schema",
            coerce_types: false,
        }
    }

    /// Should Arrow types be coerced into Parquet native types (default `false`).
    ///
    /// Setting this option to `true` will result in Parquet files that can be
    /// read by more readers, but may lose precision for Arrow types such as
    /// [`DataType::Date64`] which have no direct [corresponding Parquet type].
    ///
    /// By default, this converter does not coerce to native Parquet types. Enabling type
    /// coercion allows for meaningful representations that do not require
    /// downstream readers to consider the embedded Arrow schema, and can allow
    /// for greater compatibility with other Parquet implementations. However,
    /// type coercion also prevents data from being losslessly round-tripped.
    ///
    /// # Discussion
    ///
    /// Some Arrow types such as `Date64`, `Timestamp` and `Interval` have no
    /// corresponding Parquet logical type. Thus, they can not be losslessly
    /// round-tripped when stored using the appropriate Parquet logical type.
    /// For example, some Date64 values may be truncated when stored with
    /// parquet's native 32 bit date type.
    ///
    /// For [`List`] and [`Map`] types, some Parquet readers expect certain
    /// schema elements to have specific names (earlier versions of the spec
    /// were somewhat ambiguous on this point). Type coercion will use the names
    /// prescribed by the Parquet specification, potentially losing naming
    /// metadata from the Arrow schema.
    ///
    /// [`List`]: https://github.com/apache/parquet-format/blob/master/LogicalTypes.md#lists
    /// [`Map`]: https://github.com/apache/parquet-format/blob/master/LogicalTypes.md#maps
    /// [corresponding Parquet type]: https://github.com/apache/parquet-format/blob/master/LogicalTypes.md#date
    ///
    pub fn with_coerce_types(mut self, coerce_types: bool) -> Self {
        self.coerce_types = coerce_types;
        self
    }

    /// Set the root schema element name (defaults to `"arrow_schema"`).
    pub fn schema_root(mut self, schema_root: &'a str) -> Self {
        self.schema_root = schema_root;
        self
    }

    /// Convert the specified Arrow [`Schema`] to the desired Parquet [`SchemaDescriptor`]
    ///
    /// See example in [`ArrowSchemaConverter`]
    pub fn convert(&self, schema: &Schema) -> Result<SchemaDescriptor> {
        for field in schema.fields() {
            validate_map_key_type(field.data_type())?;
        }
        let fields = schema
            .fields()
            .iter()
            .map(|field| arrow_to_parquet_type(field, self.coerce_types).map(Arc::new))
            .collect::<Result<_>>()?;
        let group = Type::group_type_builder(self.schema_root)
            .with_fields(fields)
            .build()?;
        Ok(SchemaDescriptor::new(Arc::new(group)))
    }
}

fn parse_key_value_metadata(
    key_value_metadata: Option<&Vec<KeyValue>>,
) -> Option<HashMap<String, String>> {
    match key_value_metadata {
        Some(key_values) => {
            let map: HashMap<String, String> = key_values
                .iter()
                .filter_map(|kv| {
                    kv.value
                        .as_ref()
                        .map(|value| (kv.key.clone(), value.clone()))
                })
                .collect();

            if map.is_empty() { None } else { Some(map) }
        }
        None => None,
    }
}

/// Convert parquet column schema to arrow field.
pub fn parquet_to_arrow_field(parquet_column: &ColumnDescriptor) -> Result<Field> {
    let field = complex::convert_type(&parquet_column.self_type_ptr())?;
    let mut ret = Field::new(parquet_column.name(), field.arrow_type, field.nullable);

    let parquet_type = parquet_column.self_type();
    let basic_info = parquet_type.get_basic_info();

    let mut hash_map_size = 0;
    if basic_info.has_id() {
        hash_map_size += 1;
    }
    if has_extension_type(parquet_type) {
        hash_map_size += 1;
    }
    if hash_map_size == 0 {
        return Ok(ret);
    }
    ret.set_metadata(HashMap::with_capacity(hash_map_size));
    if basic_info.has_id() {
        ret.metadata_mut().insert(
            PARQUET_FIELD_ID_META_KEY.to_string(),
            basic_info.id().to_string(),
        );
    }
    try_add_extension_type(ret, parquet_column.self_type())
}

pub fn decimal_length_from_precision(precision: u8) -> usize {
    // digits = floor(log_10(2^(8*n - 1) - 1))  // definition in parquet's logical types
    // ceil(digits) = log10(2^(8*n - 1) - 1)
    // 10^ceil(digits) = 2^(8*n - 1) - 1
    // 10^ceil(digits) + 1 = 2^(8*n - 1)
    // log2(10^ceil(digits) + 1) = (8*n - 1)
    // log2(10^ceil(digits) + 1) + 1 = 8*n
    // (log2(10^ceil(a) + 1) + 1) / 8 = n
    (((10.0_f64.powi(precision as i32) + 1.0).log2() + 1.0) / 8.0).ceil() as usize
}

/// Convert an arrow field to a parquet `Type`
fn arrow_to_parquet_type(field: &Field, coerce_types: bool) -> Result<Type> {
    const PARQUET_LIST_ELEMENT_NAME: &str = "element";
    const PARQUET_MAP_STRUCT_NAME: &str = "key_value";
    const PARQUET_KEY_FIELD_NAME: &str = "key";
    const PARQUET_VALUE_FIELD_NAME: &str = "value";

    let name = field.name().as_str();
    let repetition = if field.is_nullable() {
        Repetition::OPTIONAL
    } else {
        Repetition::REQUIRED
    };
    let id = field_id(field);
    // create type from field
    match field.data_type() {
        DataType::Null => Type::primitive_type_builder(name, PhysicalType::INT32)
            .with_logical_type(Some(LogicalType::Unknown))
            .with_repetition(repetition)
            .with_id(id)
            .build(),
        DataType::Boolean => Type::primitive_type_builder(name, PhysicalType::BOOLEAN)
            .with_repetition(repetition)
            .with_id(id)
            .build(),
        DataType::Int8 => Type::primitive_type_builder(name, PhysicalType::INT32)
            .with_logical_type(Some(LogicalType::integer(8, true)))
            .with_repetition(repetition)
            .with_id(id)
            .build(),
        DataType::Int16 => Type::primitive_type_builder(name, PhysicalType::INT32)
            .with_logical_type(Some(LogicalType::integer(16, true)))
            .with_repetition(repetition)
            .with_id(id)
            .build(),
        DataType::Int32 => Type::primitive_type_builder(name, PhysicalType::INT32)
            .with_repetition(repetition)
            .with_id(id)
            .build(),
        DataType::Int64 => Type::primitive_type_builder(name, PhysicalType::INT64)
            .with_repetition(repetition)
            .with_id(id)
            .build(),
        DataType::UInt8 => Type::primitive_type_builder(name, PhysicalType::INT32)
            .with_logical_type(Some(LogicalType::integer(8, false)))
            .with_repetition(repetition)
            .with_id(id)
            .build(),
        DataType::UInt16 => Type::primitive_type_builder(name, PhysicalType::INT32)
            .with_logical_type(Some(LogicalType::integer(16, false)))
            .with_repetition(repetition)
            .with_id(id)
            .build(),
        DataType::UInt32 => Type::primitive_type_builder(name, PhysicalType::INT32)
            .with_logical_type(Some(LogicalType::integer(32, false)))
            .with_repetition(repetition)
            .with_id(id)
            .build(),
        DataType::UInt64 => Type::primitive_type_builder(name, PhysicalType::INT64)
            .with_logical_type(Some(LogicalType::integer(64, false)))
            .with_repetition(repetition)
            .with_id(id)
            .build(),
        DataType::Float16 => Type::primitive_type_builder(name, PhysicalType::FIXED_LEN_BYTE_ARRAY)
            .with_repetition(repetition)
            .with_id(id)
            .with_logical_type(Some(LogicalType::Float16))
            .with_length(2)
            .build(),
        DataType::Float32 => Type::primitive_type_builder(name, PhysicalType::FLOAT)
            .with_repetition(repetition)
            .with_id(id)
            .build(),
        DataType::Float64 => Type::primitive_type_builder(name, PhysicalType::DOUBLE)
            .with_repetition(repetition)
            .with_id(id)
            .build(),
        DataType::Timestamp(TimeUnit::Second, _) => {
            // Cannot represent seconds in LogicalType
            Type::primitive_type_builder(name, PhysicalType::INT64)
                .with_repetition(repetition)
                .with_id(id)
                .build()
        }
        DataType::Timestamp(time_unit, tz) => {
            Type::primitive_type_builder(name, PhysicalType::INT64)
                .with_logical_type(Some(LogicalType::timestamp(
                    // If timezone set, values are normalized to UTC timezone
                    matches!(tz, Some(z) if !z.as_ref().is_empty()),
                    match time_unit {
                        TimeUnit::Second => unreachable!(),
                        TimeUnit::Millisecond => ParquetTimeUnit::MILLIS,
                        TimeUnit::Microsecond => ParquetTimeUnit::MICROS,
                        TimeUnit::Nanosecond => ParquetTimeUnit::NANOS,
                    },
                )))
                .with_repetition(repetition)
                .with_id(id)
                .build()
        }
        DataType::Date32 => Type::primitive_type_builder(name, PhysicalType::INT32)
            .with_logical_type(Some(LogicalType::Date))
            .with_repetition(repetition)
            .with_id(id)
            .build(),
        DataType::Date64 => {
            if coerce_types {
                Type::primitive_type_builder(name, PhysicalType::INT32)
                    .with_logical_type(Some(LogicalType::Date))
                    .with_repetition(repetition)
                    .with_id(id)
                    .build()
            } else {
                Type::primitive_type_builder(name, PhysicalType::INT64)
                    .with_repetition(repetition)
                    .with_id(id)
                    .build()
            }
        }
        DataType::Time32(TimeUnit::Second) => {
            // Cannot represent seconds in LogicalType
            Type::primitive_type_builder(name, PhysicalType::INT32)
                .with_repetition(repetition)
                .with_id(id)
                .build()
        }
        DataType::Time32(unit) => Type::primitive_type_builder(name, PhysicalType::INT32)
            .with_logical_type(Some(LogicalType::time(
                field.metadata().contains_key("adjusted_to_utc"),
                match unit {
                    TimeUnit::Millisecond => ParquetTimeUnit::MILLIS,
                    u => unreachable!("Invalid unit for Time32: {:?}", u),
                },
            )))
            .with_repetition(repetition)
            .with_id(id)
            .build(),
        DataType::Time64(unit) => Type::primitive_type_builder(name, PhysicalType::INT64)
            .with_logical_type(Some(LogicalType::time(
                field.metadata().contains_key("adjusted_to_utc"),
                match unit {
                    TimeUnit::Microsecond => ParquetTimeUnit::MICROS,
                    TimeUnit::Nanosecond => ParquetTimeUnit::NANOS,
                    u => unreachable!("Invalid unit for Time64: {:?}", u),
                },
            )))
            .with_repetition(repetition)
            .with_id(id)
            .build(),
        DataType::Duration(_) => Type::primitive_type_builder(name, PhysicalType::INT64)
            .with_repetition(repetition)
            .with_id(id)
            .build(),
        DataType::Interval(_) => {
            Type::primitive_type_builder(name, PhysicalType::FIXED_LEN_BYTE_ARRAY)
                .with_converted_type(ConvertedType::INTERVAL)
                .with_repetition(repetition)
                .with_id(id)
                .with_length(12)
                .build()
        }
        DataType::Binary | DataType::LargeBinary => {
            Type::primitive_type_builder(name, PhysicalType::BYTE_ARRAY)
                .with_repetition(repetition)
                .with_id(id)
                .with_logical_type(logical_type_for_binary(field))
                .build()
        }
        DataType::FixedSizeBinary(length) => {
            Type::primitive_type_builder(name, PhysicalType::FIXED_LEN_BYTE_ARRAY)
                .with_repetition(repetition)
                .with_id(id)
                .with_length(*length)
                .with_logical_type(logical_type_for_fixed_size_binary(field))
                .build()
        }
        DataType::BinaryView => Type::primitive_type_builder(name, PhysicalType::BYTE_ARRAY)
            .with_repetition(repetition)
            .with_id(id)
            .with_logical_type(logical_type_for_binary_view(field))
            .build(),
        DataType::Decimal32(precision, scale)
        | DataType::Decimal64(precision, scale)
        | DataType::Decimal128(precision, scale)
        | DataType::Decimal256(precision, scale) => {
            // Decimal precision determines the Parquet physical type to use.
            // Following the: https://github.com/apache/parquet-format/blob/master/LogicalTypes.md#decimal
            let (physical_type, length) = if *precision >= 1 && *precision <= 9 {
                (PhysicalType::INT32, -1)
            } else if *precision <= 18 {
                (PhysicalType::INT64, -1)
            } else {
                (
                    PhysicalType::FIXED_LEN_BYTE_ARRAY,
                    decimal_length_from_precision(*precision) as i32,
                )
            };
            Type::primitive_type_builder(name, physical_type)
                .with_repetition(repetition)
                .with_id(id)
                .with_length(length)
                .with_logical_type(Some(LogicalType::decimal(*scale as i32, *precision as i32)))
                .with_precision(*precision as i32)
                .with_scale(*scale as i32)
                .build()
        }
        DataType::Utf8 | DataType::LargeUtf8 => {
            Type::primitive_type_builder(name, PhysicalType::BYTE_ARRAY)
                .with_logical_type(logical_type_for_string(field))
                .with_repetition(repetition)
                .with_id(id)
                .build()
        }
        DataType::Utf8View => Type::primitive_type_builder(name, PhysicalType::BYTE_ARRAY)
            .with_logical_type(logical_type_for_string(field))
            .with_repetition(repetition)
            .with_id(id)
            .build(),
        DataType::List(f)
        | DataType::FixedSizeList(f, _)
        | DataType::LargeList(f)
        | DataType::ListView(f)
        | DataType::LargeListView(f) => {
            let field_ref = if coerce_types && f.name() != PARQUET_LIST_ELEMENT_NAME {
                // Ensure proper naming per the Parquet specification
                let ff = f.as_ref().clone().with_name(PARQUET_LIST_ELEMENT_NAME);
                Arc::new(arrow_to_parquet_type(&ff, coerce_types)?)
            } else {
                Arc::new(arrow_to_parquet_type(f, coerce_types)?)
            };

            Type::group_type_builder(name)
                .with_fields(vec![Arc::new(
                    Type::group_type_builder("list")
                        .with_fields(vec![field_ref])
                        .with_repetition(Repetition::REPEATED)
                        .build()?,
                )])
                .with_logical_type(Some(LogicalType::List))
                .with_repetition(repetition)
                .with_id(id)
                .build()
        }
        DataType::Struct(fields) => {
            if fields.is_empty() {
                return Err(arrow_err!("Parquet does not support writing empty structs",));
            }
            // recursively convert children to types/nodes
            let fields = fields
                .iter()
                .map(|f| arrow_to_parquet_type(f, coerce_types).map(Arc::new))
                .collect::<Result<_>>()?;
            Type::group_type_builder(name)
                .with_fields(fields)
                .with_repetition(repetition)
                .with_id(id)
                .with_logical_type(logical_type_for_struct(field))
                .build()
        }
        DataType::Map(field, _) => {
            if let DataType::Struct(struct_fields) = field.data_type() {
                // If coercing then set inner struct name to "key_value"
                let map_struct_name = if coerce_types {
                    PARQUET_MAP_STRUCT_NAME
                } else {
                    field.name()
                };

                // If coercing then ensure struct fields are named "key" and "value"
                let fix_map_field = |name: &str, fld: &Arc<Field>| -> Result<Arc<Type>> {
                    if coerce_types && fld.name() != name {
                        let f = fld.as_ref().clone().with_name(name);
                        Ok(Arc::new(arrow_to_parquet_type(&f, coerce_types)?))
                    } else {
                        Ok(Arc::new(arrow_to_parquet_type(fld, coerce_types)?))
                    }
                };
                let key_field = fix_map_field(PARQUET_KEY_FIELD_NAME, &struct_fields[0])?;
                let val_field = fix_map_field(PARQUET_VALUE_FIELD_NAME, &struct_fields[1])?;

                Type::group_type_builder(name)
                    .with_fields(vec![Arc::new(
                        Type::group_type_builder(map_struct_name)
                            .with_fields(vec![key_field, val_field])
                            .with_repetition(Repetition::REPEATED)
                            .build()?,
                    )])
                    .with_logical_type(Some(LogicalType::Map))
                    .with_repetition(repetition)
                    .with_id(id)
                    .build()
            } else {
                Err(arrow_err!(
                    "DataType::Map should contain a struct field child",
                ))
            }
        }
        DataType::Union(_, _) => unimplemented!("See ARROW-8817."),
        DataType::Dictionary(_, value) => {
            // Dictionary encoding not handled at the schema level
            let dict_field = field.clone().with_data_type(value.as_ref().clone());
            arrow_to_parquet_type(&dict_field, coerce_types)
        }
        DataType::RunEndEncoded(_, value) => {
            // Run-end encoding is a physical layout. The Parquet column uses the
            // run value type and is optional when either the run array or its
            // values are nullable, so schema conversion recurses into the value
            // field here.
            let value_field = field
                .clone()
                .with_data_type(value.data_type().clone())
                .with_nullable(field.is_nullable() || value.is_nullable());
            arrow_to_parquet_type(&value_field, coerce_types)
        }
    }
}

fn field_id(field: &Field) -> Option<i32> {
    let value = field.metadata().get(super::PARQUET_FIELD_ID_META_KEY)?;
    value.parse().ok() // Fail quietly if not a valid integer
}

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

    use std::{collections::HashMap, sync::Arc};

    use crate::arrow::PARQUET_FIELD_ID_META_KEY;
    use crate::file::metadata::KeyValue;
    use crate::file::reader::FileReader;
    use crate::{
        arrow::{ArrowWriter, arrow_reader::ParquetRecordBatchReaderBuilder},
        schema::{parser::parse_message_type, types::SchemaDescriptor},
    };
    use arrow::datatypes::{DataType, Field, IntervalUnit, TimeUnit};

    #[test]
    fn test_flat_primitives() {
        let message_type = "
        message test_schema {
            REQUIRED BOOLEAN boolean;
            REQUIRED INT32   int8  (INT_8);
            REQUIRED INT32   int16 (INT_16);
            REQUIRED INT32   uint8 (INTEGER(8,false));
            REQUIRED INT32   uint16 (INTEGER(16,false));
            REQUIRED INT32   int32;
            REQUIRED INT64   int64;
            OPTIONAL DOUBLE  double;
            OPTIONAL FLOAT   float;
            OPTIONAL FIXED_LEN_BYTE_ARRAY (2) float16 (FLOAT16);
            OPTIONAL BINARY  string (UTF8);
            OPTIONAL BINARY  string_2 (STRING);
            OPTIONAL BINARY  json (JSON);
        }
        ";
        let parquet_group_type = parse_message_type(message_type).unwrap();

        let parquet_schema = SchemaDescriptor::new(Arc::new(parquet_group_type));
        let converted_arrow_schema = parquet_to_arrow_schema(&parquet_schema, None).unwrap();

        let arrow_fields = Fields::from(vec![
            Field::new("boolean", DataType::Boolean, false),
            Field::new("int8", DataType::Int8, false),
            Field::new("int16", DataType::Int16, false),
            Field::new("uint8", DataType::UInt8, false),
            Field::new("uint16", DataType::UInt16, false),
            Field::new("int32", DataType::Int32, false),
            Field::new("int64", DataType::Int64, false),
            Field::new("double", DataType::Float64, true),
            Field::new("float", DataType::Float32, true),
            Field::new("float16", DataType::Float16, true),
            Field::new("string", DataType::Utf8, true),
            Field::new("string_2", DataType::Utf8, true),
            json_field(),
        ]);

        assert_eq!(&arrow_fields, converted_arrow_schema.fields());
    }

    /// Return the expected Field for a Parquet column annotated with
    /// the JSON logical type.
    fn json_field() -> Field {
        #[cfg(feature = "arrow_canonical_extension_types")]
        {
            Field::new("json", DataType::Utf8, true)
                .with_extension_type(arrow_schema::extension::Json::default())
        }
        #[cfg(not(feature = "arrow_canonical_extension_types"))]
        {
            Field::new("json", DataType::Utf8, true)
        }
    }

    #[test]
    fn test_decimal_fields() {
        let message_type = "
        message test_schema {
                    REQUIRED INT32 decimal1 (DECIMAL(4,2));
                    REQUIRED INT64 decimal2 (DECIMAL(12,2));
                    REQUIRED FIXED_LEN_BYTE_ARRAY (16) decimal3 (DECIMAL(30,2));
                    REQUIRED BYTE_ARRAY decimal4 (DECIMAL(33,2));
                    REQUIRED BYTE_ARRAY decimal5 (DECIMAL(38,2));
                    REQUIRED FIXED_LEN_BYTE_ARRAY (17) decimal6 (DECIMAL(39,2));
                    REQUIRED BYTE_ARRAY decimal7 (DECIMAL(39,2));
        }
        ";

        let parquet_group_type = parse_message_type(message_type).unwrap();

        let parquet_schema = SchemaDescriptor::new(Arc::new(parquet_group_type));
        let converted_arrow_schema = parquet_to_arrow_schema(&parquet_schema, None).unwrap();

        let arrow_fields = Fields::from(vec![
            Field::new("decimal1", DataType::Decimal128(4, 2), false),
            Field::new("decimal2", DataType::Decimal128(12, 2), false),
            Field::new("decimal3", DataType::Decimal128(30, 2), false),
            Field::new("decimal4", DataType::Decimal128(33, 2), false),
            Field::new("decimal5", DataType::Decimal128(38, 2), false),
            Field::new("decimal6", DataType::Decimal256(39, 2), false),
            Field::new("decimal7", DataType::Decimal256(39, 2), false),
        ]);
        assert_eq!(&arrow_fields, converted_arrow_schema.fields());
    }

    #[test]
    fn test_decimal_precision_physical_types() {
        let cases = [
            (1, PhysicalType::INT32, None),
            (9, PhysicalType::INT32, None),
            (10, PhysicalType::INT64, None),
            (18, PhysicalType::INT64, None),
            (19, PhysicalType::FIXED_LEN_BYTE_ARRAY, Some(9)),
            (38, PhysicalType::FIXED_LEN_BYTE_ARRAY, Some(16)),
        ];

        for (precision, expected_type, expected_length) in cases {
            let arrow_schema = Schema::new(vec![Field::new(
                "decimal",
                DataType::Decimal128(precision, 0),
                false,
            )]);
            let parquet_schema = ArrowSchemaConverter::new().convert(&arrow_schema).unwrap();
            let column = parquet_schema.column(0);

            assert_eq!(
                column.physical_type(),
                expected_type,
                "precision {precision}"
            );
            if let Some(expected_length) = expected_length {
                assert_eq!(
                    column.type_length(),
                    expected_length,
                    "precision {precision}"
                );
            }
        }
    }

    #[test]
    fn test_byte_array_fields() {
        let message_type = "
        message test_schema {
            REQUIRED BYTE_ARRAY binary;
            REQUIRED FIXED_LEN_BYTE_ARRAY (20) fixed_binary;
        }
        ";

        let parquet_group_type = parse_message_type(message_type).unwrap();

        let parquet_schema = SchemaDescriptor::new(Arc::new(parquet_group_type));
        let converted_arrow_schema = parquet_to_arrow_schema(&parquet_schema, None).unwrap();

        let arrow_fields = Fields::from(vec![
            Field::new("binary", DataType::Binary, false),
            Field::new("fixed_binary", DataType::FixedSizeBinary(20), false),
        ]);
        assert_eq!(&arrow_fields, converted_arrow_schema.fields());
    }

    #[test]
    fn test_duplicate_fields() {
        let message_type = "
        message test_schema {
            REQUIRED BOOLEAN boolean;
            REQUIRED INT32 int8 (INT_8);
        }
        ";

        let parquet_group_type = parse_message_type(message_type).unwrap();

        let parquet_schema = SchemaDescriptor::new(Arc::new(parquet_group_type));
        let converted_arrow_schema = parquet_to_arrow_schema(&parquet_schema, None).unwrap();

        let arrow_fields = Fields::from(vec![
            Field::new("boolean", DataType::Boolean, false),
            Field::new("int8", DataType::Int8, false),
        ]);
        assert_eq!(&arrow_fields, converted_arrow_schema.fields());

        let converted_arrow_schema =
            parquet_to_arrow_schema_by_columns(&parquet_schema, ProjectionMask::all(), None)
                .unwrap();
        assert_eq!(&arrow_fields, converted_arrow_schema.fields());
    }

    #[test]
    fn test_parquet_lists() {
        let mut arrow_fields = Vec::new();

        // LIST encoding example taken from parquet-format/LogicalTypes.md
        let message_type = "
        message test_schema {
          REQUIRED GROUP my_list (LIST) {
            REPEATED GROUP list {
              OPTIONAL BINARY element (UTF8);
            }
          }
          OPTIONAL GROUP my_list (LIST) {
            REPEATED GROUP list {
              REQUIRED BINARY element (UTF8);
            }
          }
          OPTIONAL GROUP array_of_arrays (LIST) {
            REPEATED GROUP list {
              REQUIRED GROUP element (LIST) {
                REPEATED GROUP list {
                  REQUIRED INT32 element;
                }
              }
            }
          }
          OPTIONAL GROUP my_list (LIST) {
            REPEATED GROUP element {
              REQUIRED BINARY str (UTF8);
            }
          }
          OPTIONAL GROUP my_list (LIST) {
            REPEATED INT32 element;
          }
          OPTIONAL GROUP my_list (LIST) {
            REPEATED GROUP element {
              REQUIRED BINARY str (UTF8);
              REQUIRED INT32 num;
            }
          }
          OPTIONAL GROUP my_list (LIST) {
            REPEATED GROUP array {
              REQUIRED BINARY str (UTF8);
            }

          }
          OPTIONAL GROUP my_list (LIST) {
            REPEATED GROUP my_list_tuple {
              REQUIRED BINARY str (UTF8);
            }
          }
          REPEATED INT32 name;
        }
        ";

        // // List<String> (list non-null, elements nullable)
        // required group my_list (LIST) {
        //   repeated group list {
        //     optional binary element (UTF8);
        //   }
        // }
        {
            arrow_fields.push(Field::new_list(
                "my_list",
                Field::new("element", DataType::Utf8, true),
                false,
            ));
        }

        // // List<String> (list nullable, elements non-null)
        // optional group my_list (LIST) {
        //   repeated group list {
        //     required binary element (UTF8);
        //   }
        // }
        {
            arrow_fields.push(Field::new_list(
                "my_list",
                Field::new("element", DataType::Utf8, false),
                true,
            ));
        }

        // Element types can be nested structures. For example, a list of lists:
        //
        // // List<List<Integer>>
        // optional group array_of_arrays (LIST) {
        //   repeated group list {
        //     required group element (LIST) {
        //       repeated group list {
        //         required int32 element;
        //       }
        //     }
        //   }
        // }
        {
            let arrow_inner_list = Field::new("element", DataType::Int32, false);
            arrow_fields.push(Field::new_list(
                "array_of_arrays",
                Field::new_list("element", arrow_inner_list, false),
                true,
            ));
        }

        // // List<String> (list nullable, elements non-null)
        // optional group my_list (LIST) {
        //   repeated group element {
        //     required binary str (UTF8);
        //   };
        // }
        {
            arrow_fields.push(Field::new_list(
                "my_list",
                Field::new("str", DataType::Utf8, false),
                true,
            ));
        }

        // // List<Integer> (nullable list, non-null elements)
        // optional group my_list (LIST) {
        //   repeated int32 element;
        // }
        {
            arrow_fields.push(Field::new_list(
                "my_list",
                Field::new("element", DataType::Int32, false),
                true,
            ));
        }

        // // List<Tuple<String, Integer>> (nullable list, non-null elements)
        // optional group my_list (LIST) {
        //   repeated group element {
        //     required binary str (UTF8);
        //     required int32 num;
        //   };
        // }
        {
            let fields = vec![
                Field::new("str", DataType::Utf8, false),
                Field::new("num", DataType::Int32, false),
            ];
            arrow_fields.push(Field::new_list(
                "my_list",
                Field::new_struct("element", fields, false),
                true,
            ));
        }

        // // List<OneTuple<String>> (nullable list, non-null elements)
        // optional group my_list (LIST) {
        //   repeated group array {
        //     required binary str (UTF8);
        //   };
        // }
        // Special case: group is named array
        {
            let fields = vec![Field::new("str", DataType::Utf8, false)];
            arrow_fields.push(Field::new_list(
                "my_list",
                Field::new_struct("array", fields, false),
                true,
            ));
        }

        // // List<OneTuple<String>> (nullable list, non-null elements)
        // optional group my_list (LIST) {
        //   repeated group my_list_tuple {
        //     required binary str (UTF8);
        //   };
        // }
        // Special case: group named ends in _tuple
        {
            let fields = vec![Field::new("str", DataType::Utf8, false)];
            arrow_fields.push(Field::new_list(
                "my_list",
                Field::new_struct("my_list_tuple", fields, false),
                true,
            ));
        }

        // One-level encoding: Only allows required lists with required cells
        //   repeated value_type name
        {
            arrow_fields.push(Field::new_list(
                "name",
                Field::new("name", DataType::Int32, false),
                false,
            ));
        }

        let parquet_group_type = parse_message_type(message_type).unwrap();

        let parquet_schema = SchemaDescriptor::new(Arc::new(parquet_group_type));
        let converted_arrow_schema = parquet_to_arrow_schema(&parquet_schema, None).unwrap();
        let converted_fields = converted_arrow_schema.fields();

        assert_eq!(arrow_fields.len(), converted_fields.len());
        for i in 0..arrow_fields.len() {
            assert_eq!(&arrow_fields[i], converted_fields[i].as_ref(), "{i}");
        }
    }

    #[test]
    fn test_parquet_list_nullable() {
        let mut arrow_fields = Vec::new();

        let message_type = "
        message test_schema {
          REQUIRED GROUP my_list1 (LIST) {
            REPEATED GROUP list {
              OPTIONAL BINARY element (UTF8);
            }
          }
          OPTIONAL GROUP my_list2 (LIST) {
            REPEATED GROUP list {
              REQUIRED BINARY element (UTF8);
            }
          }
          REQUIRED GROUP my_list3 (LIST) {
            REPEATED GROUP list {
              REQUIRED BINARY element (UTF8);
            }
          }
        }
        ";

        // // List<String> (list non-null, elements nullable)
        // required group my_list1 (LIST) {
        //   repeated group list {
        //     optional binary element (UTF8);
        //   }
        // }
        {
            arrow_fields.push(Field::new_list(
                "my_list1",
                Field::new("element", DataType::Utf8, true),
                false,
            ));
        }

        // // List<String> (list nullable, elements non-null)
        // optional group my_list2 (LIST) {
        //   repeated group list {
        //     required binary element (UTF8);
        //   }
        // }
        {
            arrow_fields.push(Field::new_list(
                "my_list2",
                Field::new("element", DataType::Utf8, false),
                true,
            ));
        }

        // // List<String> (list non-null, elements non-null)
        // repeated group my_list3 (LIST) {
        //   repeated group list {
        //     required binary element (UTF8);
        //   }
        // }
        {
            arrow_fields.push(Field::new_list(
                "my_list3",
                Field::new("element", DataType::Utf8, false),
                false,
            ));
        }

        let parquet_group_type = parse_message_type(message_type).unwrap();

        let parquet_schema = SchemaDescriptor::new(Arc::new(parquet_group_type));
        let converted_arrow_schema = parquet_to_arrow_schema(&parquet_schema, None).unwrap();
        let converted_fields = converted_arrow_schema.fields();

        assert_eq!(arrow_fields.len(), converted_fields.len());
        for i in 0..arrow_fields.len() {
            assert_eq!(&arrow_fields[i], converted_fields[i].as_ref());
        }
    }

    #[test]
    fn test_parquet_maps() {
        let mut arrow_fields = Vec::new();

        // LIST encoding example taken from parquet-format/LogicalTypes.md
        let message_type = "
        message test_schema {
          REQUIRED group my_map1 (MAP) {
            REPEATED group key_value {
              REQUIRED binary key (UTF8);
              OPTIONAL int32 value;
            }
          }
          OPTIONAL group my_map2 (MAP) {
            REPEATED group map {
              REQUIRED binary str (UTF8);
              REQUIRED int32 num;
            }
          }
          OPTIONAL group my_map3 (MAP_KEY_VALUE) {
            REPEATED group map {
              REQUIRED binary key (UTF8);
              OPTIONAL int32 value;
            }
          }
          REQUIRED group my_map4 (MAP) {
            REPEATED group map {
              OPTIONAL binary key (UTF8);
              REQUIRED int32 value;
            }
          }
        }
        ";

        // // Map<String, Integer>
        // required group my_map (MAP) {
        //   repeated group key_value {
        //     required binary key (UTF8);
        //     optional int32 value;
        //   }
        // }
        {
            arrow_fields.push(Field::new_map(
                "my_map1",
                "key_value",
                Field::new("key", DataType::Utf8, false),
                Field::new("value", DataType::Int32, true),
                false,
                false,
            ));
        }

        // // Map<String, Integer> (nullable map, non-null values)
        // optional group my_map (MAP) {
        //   repeated group map {
        //     required binary str (UTF8);
        //     required int32 num;
        //   }
        // }
        {
            arrow_fields.push(Field::new_map(
                "my_map2",
                "map",
                Field::new("str", DataType::Utf8, false),
                Field::new("num", DataType::Int32, false),
                false,
                true,
            ));
        }

        // // Map<String, Integer> (nullable map, nullable values)
        // optional group my_map (MAP_KEY_VALUE) {
        //   repeated group map {
        //     required binary key (UTF8);
        //     optional int32 value;
        //   }
        // }
        {
            arrow_fields.push(Field::new_map(
                "my_map3",
                "map",
                Field::new("key", DataType::Utf8, false),
                Field::new("value", DataType::Int32, true),
                false,
                true,
            ));
        }

        // // Map<String, Integer> (non-compliant nullable key)
        // group my_map (MAP_KEY_VALUE) {
        //   repeated group map {
        //     optional binary key (UTF8);
        //     required int32 value;
        //   }
        // }
        {
            arrow_fields.push(Field::new_map(
                "my_map4",
                "map",
                Field::new("key", DataType::Utf8, false), // The key is always non-nullable (#5630)
                Field::new("value", DataType::Int32, false),
                false,
                false,
            ));
        }

        let parquet_group_type = parse_message_type(message_type).unwrap();

        let parquet_schema = SchemaDescriptor::new(Arc::new(parquet_group_type));
        let converted_arrow_schema = parquet_to_arrow_schema(&parquet_schema, None).unwrap();
        let converted_fields = converted_arrow_schema.fields();

        assert_eq!(arrow_fields.len(), converted_fields.len());
        for i in 0..arrow_fields.len() {
            assert_eq!(&arrow_fields[i], converted_fields[i].as_ref());
        }
    }

    #[test]
    fn test_nested_schema() {
        let mut arrow_fields = Vec::new();
        {
            let group1_fields = Fields::from(vec![
                Field::new("leaf1", DataType::Boolean, false),
                Field::new("leaf2", DataType::Int32, false),
            ]);
            let group1_struct = Field::new("group1", DataType::Struct(group1_fields), false);
            arrow_fields.push(group1_struct);

            let leaf3_field = Field::new("leaf3", DataType::Int64, false);
            arrow_fields.push(leaf3_field);
        }

        let message_type = "
        message test_schema {
          REQUIRED GROUP group1 {
            REQUIRED BOOLEAN leaf1;
            REQUIRED INT32 leaf2;
          }
          REQUIRED INT64 leaf3;
        }
        ";
        let parquet_group_type = parse_message_type(message_type).unwrap();

        let parquet_schema = SchemaDescriptor::new(Arc::new(parquet_group_type));
        let converted_arrow_schema = parquet_to_arrow_schema(&parquet_schema, None).unwrap();
        let converted_fields = converted_arrow_schema.fields();

        assert_eq!(arrow_fields.len(), converted_fields.len());
        for i in 0..arrow_fields.len() {
            assert_eq!(&arrow_fields[i], converted_fields[i].as_ref());
        }
    }

    #[test]
    fn test_nested_schema_partial() {
        let mut arrow_fields = Vec::new();
        {
            let group1_fields = vec![Field::new("leaf1", DataType::Int64, false)].into();
            let group1 = Field::new("group1", DataType::Struct(group1_fields), false);
            arrow_fields.push(group1);

            let group2_fields = vec![Field::new("leaf4", DataType::Int64, false)].into();
            let group2 = Field::new("group2", DataType::Struct(group2_fields), false);
            arrow_fields.push(group2);

            arrow_fields.push(Field::new("leaf5", DataType::Int64, false));
        }

        let message_type = "
        message test_schema {
          REQUIRED GROUP group1 {
            REQUIRED INT64 leaf1;
            REQUIRED INT64 leaf2;
          }
          REQUIRED  GROUP group2 {
            REQUIRED INT64 leaf3;
            REQUIRED INT64 leaf4;
          }
          REQUIRED INT64 leaf5;
        }
        ";
        let parquet_group_type = parse_message_type(message_type).unwrap();

        // Expected partial arrow schema (columns 0, 3, 4):
        // required group group1 {
        //   required int64 leaf1;
        // }
        // required group group2 {
        //   required int64 leaf4;
        // }
        // required int64 leaf5;

        let parquet_schema = SchemaDescriptor::new(Arc::new(parquet_group_type));
        let mask = ProjectionMask::leaves(&parquet_schema, [3, 0, 4, 4]);
        let converted_arrow_schema =
            parquet_to_arrow_schema_by_columns(&parquet_schema, mask, None).unwrap();
        let converted_fields = converted_arrow_schema.fields();

        assert_eq!(arrow_fields.len(), converted_fields.len());
        for i in 0..arrow_fields.len() {
            assert_eq!(&arrow_fields[i], converted_fields[i].as_ref());
        }
    }

    #[test]
    fn test_nested_schema_partial_ordering() {
        let mut arrow_fields = Vec::new();
        {
            let group1_fields = vec![Field::new("leaf1", DataType::Int64, false)].into();
            let group1 = Field::new("group1", DataType::Struct(group1_fields), false);
            arrow_fields.push(group1);

            let group2_fields = vec![Field::new("leaf4", DataType::Int64, false)].into();
            let group2 = Field::new("group2", DataType::Struct(group2_fields), false);
            arrow_fields.push(group2);

            arrow_fields.push(Field::new("leaf5", DataType::Int64, false));
        }

        let message_type = "
        message test_schema {
          REQUIRED GROUP group1 {
            REQUIRED INT64 leaf1;
            REQUIRED INT64 leaf2;
          }
          REQUIRED  GROUP group2 {
            REQUIRED INT64 leaf3;
            REQUIRED INT64 leaf4;
          }
          REQUIRED INT64 leaf5;
        }
        ";
        let parquet_group_type = parse_message_type(message_type).unwrap();

        // Expected partial arrow schema (columns 3, 4, 0):
        // required group group1 {
        //   required int64 leaf1;
        // }
        // required group group2 {
        //   required int64 leaf4;
        // }
        // required int64 leaf5;

        let parquet_schema = SchemaDescriptor::new(Arc::new(parquet_group_type));
        let mask = ProjectionMask::leaves(&parquet_schema, [3, 0, 4]);
        let converted_arrow_schema =
            parquet_to_arrow_schema_by_columns(&parquet_schema, mask, None).unwrap();
        let converted_fields = converted_arrow_schema.fields();

        assert_eq!(arrow_fields.len(), converted_fields.len());
        for i in 0..arrow_fields.len() {
            assert_eq!(&arrow_fields[i], converted_fields[i].as_ref());
        }

        let mask =
            ProjectionMask::columns(&parquet_schema, ["group2.leaf4", "group1.leaf1", "leaf5"]);
        let converted_arrow_schema =
            parquet_to_arrow_schema_by_columns(&parquet_schema, mask, None).unwrap();
        let converted_fields = converted_arrow_schema.fields();

        assert_eq!(arrow_fields.len(), converted_fields.len());
        for i in 0..arrow_fields.len() {
            assert_eq!(&arrow_fields[i], converted_fields[i].as_ref());
        }
    }

    #[test]
    fn test_repeated_nested_schema() {
        let mut arrow_fields = Vec::new();
        {
            arrow_fields.push(Field::new("leaf1", DataType::Int32, true));

            let inner_group_list = Field::new_list(
                "innerGroup",
                Field::new_struct(
                    "innerGroup",
                    vec![Field::new("leaf3", DataType::Int32, true)],
                    false,
                ),
                false,
            );

            let outer_group_list = Field::new_list(
                "outerGroup",
                Field::new_struct(
                    "outerGroup",
                    vec![Field::new("leaf2", DataType::Int32, true), inner_group_list],
                    false,
                ),
                false,
            );
            arrow_fields.push(outer_group_list);
        }

        let message_type = "
        message test_schema {
          OPTIONAL INT32 leaf1;
          REPEATED GROUP outerGroup {
            OPTIONAL INT32 leaf2;
            REPEATED GROUP innerGroup {
              OPTIONAL INT32 leaf3;
            }
          }
        }
        ";
        let parquet_group_type = parse_message_type(message_type).unwrap();

        let parquet_schema = SchemaDescriptor::new(Arc::new(parquet_group_type));
        let converted_arrow_schema = parquet_to_arrow_schema(&parquet_schema, None).unwrap();
        let converted_fields = converted_arrow_schema.fields();

        assert_eq!(arrow_fields.len(), converted_fields.len());
        for i in 0..arrow_fields.len() {
            assert_eq!(&arrow_fields[i], converted_fields[i].as_ref());
        }
    }

    #[test]
    fn test_column_desc_to_field() {
        let message_type = "
        message test_schema {
            REQUIRED BOOLEAN boolean;
            REQUIRED INT32   int8  (INT_8);
            REQUIRED INT32   uint8 (INTEGER(8,false));
            REQUIRED INT32   int16 (INT_16);
            REQUIRED INT32   uint16 (INTEGER(16,false));
            REQUIRED INT32   int32;
            REQUIRED INT64   int64;
            OPTIONAL DOUBLE  double;
            OPTIONAL FLOAT   float;
            OPTIONAL FIXED_LEN_BYTE_ARRAY (2) float16 (FLOAT16);
            OPTIONAL BINARY  string (UTF8);
            REPEATED BOOLEAN bools;
            OPTIONAL INT32   date       (DATE);
            OPTIONAL INT32   time_milli (TIME_MILLIS);
            OPTIONAL INT64   time_micro (TIME_MICROS);
            OPTIONAL INT64   time_nano (TIME(NANOS,false));
            OPTIONAL INT64   ts_milli (TIMESTAMP_MILLIS);
            REQUIRED INT64   ts_micro (TIMESTAMP_MICROS);
            REQUIRED INT64   ts_nano (TIMESTAMP(NANOS,true));
            REPEATED INT32   int_list;
            REPEATED BINARY  byte_list;
            REPEATED BINARY  string_list (UTF8);
            REQUIRED INT32 decimal_int32 (DECIMAL(8,2));
            REQUIRED INT64 decimal_int64 (DECIMAL(16,2));
            REQUIRED FIXED_LEN_BYTE_ARRAY (13) decimal_fix_length (DECIMAL(30,2));
        }
        ";
        let parquet_group_type = parse_message_type(message_type).unwrap();

        let parquet_schema = SchemaDescriptor::new(Arc::new(parquet_group_type));
        let converted_arrow_fields = parquet_schema
            .columns()
            .iter()
            .map(|c| parquet_to_arrow_field(c).unwrap())
            .collect::<Vec<Field>>();

        let arrow_fields = vec![
            Field::new("boolean", DataType::Boolean, false),
            Field::new("int8", DataType::Int8, false),
            Field::new("uint8", DataType::UInt8, false),
            Field::new("int16", DataType::Int16, false),
            Field::new("uint16", DataType::UInt16, false),
            Field::new("int32", DataType::Int32, false),
            Field::new("int64", DataType::Int64, false),
            Field::new("double", DataType::Float64, true),
            Field::new("float", DataType::Float32, true),
            Field::new("float16", DataType::Float16, true),
            Field::new("string", DataType::Utf8, true),
            Field::new_list(
                "bools",
                Field::new("bools", DataType::Boolean, false),
                false,
            ),
            Field::new("date", DataType::Date32, true),
            Field::new("time_milli", DataType::Time32(TimeUnit::Millisecond), true),
            Field::new("time_micro", DataType::Time64(TimeUnit::Microsecond), true),
            Field::new("time_nano", DataType::Time64(TimeUnit::Nanosecond), true),
            Field::new(
                "ts_milli",
                DataType::Timestamp(TimeUnit::Millisecond, Some("UTC".into())),
                true,
            ),
            Field::new(
                "ts_micro",
                DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into())),
                false,
            ),
            Field::new(
                "ts_nano",
                DataType::Timestamp(TimeUnit::Nanosecond, Some("UTC".into())),
                false,
            ),
            Field::new_list(
                "int_list",
                Field::new("int_list", DataType::Int32, false),
                false,
            ),
            Field::new_list(
                "byte_list",
                Field::new("byte_list", DataType::Binary, false),
                false,
            ),
            Field::new_list(
                "string_list",
                Field::new("string_list", DataType::Utf8, false),
                false,
            ),
            Field::new("decimal_int32", DataType::Decimal128(8, 2), false),
            Field::new("decimal_int64", DataType::Decimal128(16, 2), false),
            Field::new("decimal_fix_length", DataType::Decimal128(30, 2), false),
        ];

        assert_eq!(arrow_fields, converted_arrow_fields);
    }

    #[test]
    fn test_coerced_map_list() {
        // Create Arrow schema with non-Parquet naming
        let arrow_fields = vec![
            Field::new_list(
                "my_list",
                Field::new("item", DataType::Boolean, true),
                false,
            ),
            Field::new_map(
                "my_map",
                "my_entries",
                Field::new("my_keys", DataType::Utf8, false),
                Field::new("my_values", DataType::Int32, true),
                false,
                true,
            ),
        ];
        let arrow_schema = Schema::new(arrow_fields);

        // Create Parquet schema with coerced names
        let message_type = "
        message parquet_schema {
            REQUIRED GROUP my_list (LIST) {
                REPEATED GROUP list {
                    OPTIONAL BOOLEAN element;
                }
            }
            OPTIONAL GROUP my_map (MAP) {
                REPEATED GROUP key_value {
                    REQUIRED BINARY key (STRING);
                    OPTIONAL INT32 value;
                }
            }
        }
        ";
        let parquet_group_type = parse_message_type(message_type).unwrap();
        let parquet_schema = SchemaDescriptor::new(Arc::new(parquet_group_type));
        let converted_arrow_schema = ArrowSchemaConverter::new()
            .with_coerce_types(true)
            .convert(&arrow_schema)
            .unwrap();
        assert_eq!(
            parquet_schema.columns().len(),
            converted_arrow_schema.columns().len()
        );

        // Create Parquet schema without coerced names
        let message_type = "
        message parquet_schema {
            REQUIRED GROUP my_list (LIST) {
                REPEATED GROUP list {
                    OPTIONAL BOOLEAN item;
                }
            }
            OPTIONAL GROUP my_map (MAP) {
                REPEATED GROUP my_entries {
                    REQUIRED BINARY my_keys (STRING);
                    OPTIONAL INT32 my_values;
                }
            }
        }
        ";
        let parquet_group_type = parse_message_type(message_type).unwrap();
        let parquet_schema = SchemaDescriptor::new(Arc::new(parquet_group_type));
        let converted_arrow_schema = ArrowSchemaConverter::new()
            .with_coerce_types(false)
            .convert(&arrow_schema)
            .unwrap();
        assert_eq!(
            parquet_schema.columns().len(),
            converted_arrow_schema.columns().len()
        );
    }

    #[test]
    fn test_field_to_column_desc() {
        let message_type = "
        message arrow_schema {
            REQUIRED BOOLEAN boolean;
            REQUIRED INT32   int8  (INT_8);
            REQUIRED INT32   int16 (INTEGER(16,true));
            REQUIRED INT32   int32;
            REQUIRED INT64   int64;
            OPTIONAL DOUBLE  double;
            OPTIONAL FLOAT   float;
            OPTIONAL FIXED_LEN_BYTE_ARRAY (2) float16 (FLOAT16);
            OPTIONAL BINARY  string (STRING);
            OPTIONAL GROUP   bools (LIST) {
                REPEATED GROUP list {
                    OPTIONAL BOOLEAN element;
                }
            }
            REQUIRED GROUP   bools_non_null (LIST) {
                REPEATED GROUP list {
                    REQUIRED BOOLEAN element;
                }
            }
            OPTIONAL INT32   date       (DATE);
            OPTIONAL INT32   time_milli (TIME(MILLIS,false));
            OPTIONAL INT32   time_milli_utc (TIME(MILLIS,true));
            OPTIONAL INT64   time_micro (TIME_MICROS);
            OPTIONAL INT64   time_micro_utc (TIME(MICROS, true));
            OPTIONAL INT64   ts_milli (TIMESTAMP_MILLIS);
            REQUIRED INT64   ts_micro (TIMESTAMP(MICROS,false));
            REQUIRED INT64   ts_seconds;
            REQUIRED INT64   ts_micro_utc (TIMESTAMP(MICROS, true));
            REQUIRED INT64   ts_millis_zero_offset (TIMESTAMP(MILLIS, true));
            REQUIRED INT64   ts_millis_zero_negative_offset (TIMESTAMP(MILLIS, true));
            REQUIRED INT64   ts_micro_non_utc (TIMESTAMP(MICROS, true));
            REQUIRED GROUP struct {
                REQUIRED BOOLEAN bools;
                REQUIRED INT32 uint32 (INTEGER(32,false));
                REQUIRED GROUP   int32 (LIST) {
                    REPEATED GROUP list {
                        OPTIONAL INT32 element;
                    }
                }
            }
            REQUIRED BINARY  dictionary_strings (STRING);
            REQUIRED INT32 decimal_int32 (DECIMAL(8,2));
            REQUIRED INT64 decimal_int64 (DECIMAL(16,2));
            REQUIRED FIXED_LEN_BYTE_ARRAY (13) decimal_fix_length (DECIMAL(30,2));
            REQUIRED FIXED_LEN_BYTE_ARRAY (16) decimal128 (DECIMAL(38,2));
            REQUIRED FIXED_LEN_BYTE_ARRAY (17) decimal256 (DECIMAL(39,2));
        }
        ";
        let parquet_group_type = parse_message_type(message_type).unwrap();

        let parquet_schema = SchemaDescriptor::new(Arc::new(parquet_group_type));

        let arrow_fields = vec![
            Field::new("boolean", DataType::Boolean, false),
            Field::new("int8", DataType::Int8, false),
            Field::new("int16", DataType::Int16, false),
            Field::new("int32", DataType::Int32, false),
            Field::new("int64", DataType::Int64, false),
            Field::new("double", DataType::Float64, true),
            Field::new("float", DataType::Float32, true),
            Field::new("float16", DataType::Float16, true),
            Field::new("string", DataType::Utf8, true),
            Field::new_list(
                "bools",
                Field::new("element", DataType::Boolean, true),
                true,
            ),
            Field::new_list(
                "bools_non_null",
                Field::new("element", DataType::Boolean, false),
                false,
            ),
            Field::new("date", DataType::Date32, true),
            Field::new("time_milli", DataType::Time32(TimeUnit::Millisecond), true),
            Field::new(
                "time_milli_utc",
                DataType::Time32(TimeUnit::Millisecond),
                true,
            )
            .with_metadata(HashMap::from_iter(vec![(
                "adjusted_to_utc".to_string(),
                String::new(),
            )])),
            Field::new("time_micro", DataType::Time64(TimeUnit::Microsecond), true),
            Field::new(
                "time_micro_utc",
                DataType::Time64(TimeUnit::Microsecond),
                true,
            )
            .with_metadata(HashMap::from_iter(vec![(
                "adjusted_to_utc".to_string(),
                String::new(),
            )])),
            Field::new(
                "ts_milli",
                DataType::Timestamp(TimeUnit::Millisecond, None),
                true,
            ),
            Field::new(
                "ts_micro",
                DataType::Timestamp(TimeUnit::Microsecond, None),
                false,
            ),
            Field::new(
                "ts_seconds",
                DataType::Timestamp(TimeUnit::Second, Some("UTC".into())),
                false,
            ),
            Field::new(
                "ts_micro_utc",
                DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into())),
                false,
            ),
            Field::new(
                "ts_millis_zero_offset",
                DataType::Timestamp(TimeUnit::Millisecond, Some("+00:00".into())),
                false,
            ),
            Field::new(
                "ts_millis_zero_negative_offset",
                DataType::Timestamp(TimeUnit::Millisecond, Some("-00:00".into())),
                false,
            ),
            Field::new(
                "ts_micro_non_utc",
                DataType::Timestamp(TimeUnit::Microsecond, Some("+01:00".into())),
                false,
            ),
            Field::new_struct(
                "struct",
                vec![
                    Field::new("bools", DataType::Boolean, false),
                    Field::new("uint32", DataType::UInt32, false),
                    Field::new_list("int32", Field::new("element", DataType::Int32, true), false),
                ],
                false,
            ),
            Field::new_dictionary("dictionary_strings", DataType::Int32, DataType::Utf8, false),
            Field::new("decimal_int32", DataType::Decimal128(8, 2), false),
            Field::new("decimal_int64", DataType::Decimal128(16, 2), false),
            Field::new("decimal_fix_length", DataType::Decimal128(30, 2), false),
            Field::new("decimal128", DataType::Decimal128(38, 2), false),
            Field::new("decimal256", DataType::Decimal256(39, 2), false),
        ];
        let arrow_schema = Schema::new(arrow_fields);
        let converted_arrow_schema = ArrowSchemaConverter::new().convert(&arrow_schema).unwrap();

        assert_eq!(
            parquet_schema.columns().len(),
            converted_arrow_schema.columns().len()
        );
        parquet_schema
            .columns()
            .iter()
            .zip(converted_arrow_schema.columns())
            .for_each(|(a, b)| {
                // Only check logical type if it's set on the Parquet side.
                // This is because the Arrow conversion always sets logical type,
                // even if there wasn't originally one.
                // This is not an issue, but is an inconvenience for this test.
                match a.logical_type_ref() {
                    Some(_) => {
                        assert_eq!(a, b)
                    }
                    None => {
                        assert_eq!(a.name(), b.name());
                        assert_eq!(a.physical_type(), b.physical_type());
                        assert_eq!(a.converted_type(), b.converted_type());
                    }
                }
            });
    }

    #[test]
    #[should_panic(expected = "Parquet does not support writing empty structs")]
    fn test_empty_struct_field() {
        let arrow_fields = vec![Field::new(
            "struct",
            DataType::Struct(Fields::empty()),
            false,
        )];
        let arrow_schema = Schema::new(arrow_fields);
        let converted_arrow_schema = ArrowSchemaConverter::new()
            .with_coerce_types(true)
            .convert(&arrow_schema);

        converted_arrow_schema.unwrap();
    }

    #[test]
    fn test_run_end_schema_conversion_recurses_into_value_type() {
        fn run_end_type(value_type: DataType, nullable_values: bool) -> DataType {
            DataType::RunEndEncoded(
                Arc::new(Field::new("run_ends", DataType::Int32, false)),
                Arc::new(Field::new("values", value_type, nullable_values)),
            )
        }

        // A leaf REE contributes the value's physical type and combined
        // nullability to the Parquet schema.
        let leaf_schema = Schema::new(vec![Field::new(
            "ree",
            run_end_type(DataType::Int32, true),
            false,
        )]);
        let converted = ArrowSchemaConverter::new().convert(&leaf_schema).unwrap();
        assert_eq!(converted.columns().len(), 1);
        assert_eq!(converted.column(0).name(), "ree");
        assert_eq!(converted.column(0).physical_type(), PhysicalType::INT32);
        assert_eq!(converted.column(0).max_def_level(), 1);

        // Non-leaf value types recurse into the value type rather than
        // exposing Arrow's physical run-end wrapper in the Parquet schema.
        let list_schema = Schema::new(vec![Field::new(
            "ree_list",
            run_end_type(
                DataType::List(Arc::new(Field::new("element", DataType::Int32, true))),
                true,
            ),
            true,
        )]);
        let converted = ArrowSchemaConverter::new().convert(&list_schema).unwrap();
        assert_eq!(converted.columns().len(), 1);
        assert_eq!(converted.column(0).physical_type(), PhysicalType::INT32);
        assert!(
            converted.column(0).max_rep_level() >= 1,
            "list element should be repeated"
        );

        let dictionary_schema = Schema::new(vec![Field::new(
            "ree_dictionary",
            run_end_type(
                DataType::Dictionary(Box::new(DataType::Int8), Box::new(DataType::Utf8)),
                true,
            ),
            true,
        )]);
        let converted = ArrowSchemaConverter::new()
            .convert(&dictionary_schema)
            .unwrap();
        assert_eq!(converted.columns().len(), 1);
        assert_eq!(
            converted.column(0).physical_type(),
            PhysicalType::BYTE_ARRAY
        );
    }

    fn map_type_with_key(key: Field) -> DataType {
        DataType::Map(
            Arc::new(Field::new(
                "entries",
                DataType::Struct(vec![key, Field::new("value", DataType::Int32, true)].into()),
                false,
            )),
            false,
        )
    }

    #[test]
    fn test_map_key_schema_rejects_logically_nullable_keys() {
        let ree = DataType::RunEndEncoded(
            Arc::new(Field::new("run_ends", DataType::Int32, false)),
            Arc::new(Field::new("values", DataType::Int32, true)),
        );
        let dictionary_ree = DataType::Dictionary(Box::new(DataType::Int8), Box::new(ree.clone()));
        let nested_ree = DataType::RunEndEncoded(
            Arc::new(Field::new("run_ends", DataType::Int32, false)),
            Arc::new(Field::new("values", dictionary_ree.clone(), false)),
        );
        for key in [
            Field::new("key", DataType::Int32, true),
            Field::new("key", ree, false),
            Field::new("key", dictionary_ree, false),
            Field::new("key", nested_ree, false),
        ] {
            let map = map_type_with_key(key);
            for data_type in [
                map.clone(),
                DataType::Struct(vec![Field::new("nested", map.clone(), true)].into()),
                DataType::List(Arc::new(Field::new("item", map.clone(), true))),
                DataType::Dictionary(Box::new(DataType::Int8), Box::new(map.clone())),
                DataType::RunEndEncoded(
                    Arc::new(Field::new("run_ends", DataType::Int32, false)),
                    Arc::new(Field::new("values", map, true)),
                ),
            ] {
                let schema = Schema::new(vec![Field::new("map", data_type, true)]);
                for coerce_types in [false, true] {
                    let err = ArrowSchemaConverter::new()
                        .with_coerce_types(coerce_types)
                        .convert(&schema)
                        .unwrap_err();
                    assert!(
                        err.to_string()
                            .contains("Map key field 'key' must be non-nullable"),
                        "{err}"
                    );
                }
            }
        }
    }

    #[test]
    fn test_map_key_schema_preserves_required_keys_and_nullable_members() {
        for key_type in [
            DataType::Int32,
            DataType::Dictionary(Box::new(DataType::Int8), Box::new(DataType::Int32)),
            DataType::RunEndEncoded(
                Arc::new(Field::new("run_ends", DataType::Int32, false)),
                Arc::new(Field::new("values", DataType::Int32, false)),
            ),
            DataType::Struct(vec![Field::new("member", DataType::Int32, true)].into()),
        ] {
            let schema = Schema::new(vec![Field::new(
                "map",
                map_type_with_key(Field::new("key", key_type, false)),
                true,
            )]);
            let parquet = ArrowSchemaConverter::new().convert(&schema).unwrap();
            let key = &parquet.root_schema().get_fields()[0].get_fields()[0].get_fields()[0];
            assert_eq!(key.get_basic_info().repetition(), Repetition::REQUIRED);

            let mut props = WriterProperties::builder().build();
            add_encoded_arrow_schema_to_metadata(&schema, &mut props);
            let encoded = props
                .key_value_metadata()
                .unwrap()
                .iter()
                .find(|kv| kv.key == crate::arrow::ARROW_SCHEMA_META_KEY)
                .unwrap()
                .value
                .as_deref()
                .unwrap();
            let decoded = get_arrow_schema_from_metadata(encoded).unwrap();
            let DataType::Map(entries, _) = decoded.field(0).data_type() else {
                unreachable!()
            };
            let DataType::Struct(fields) = entries.data_type() else {
                unreachable!()
            };
            assert!(!fields[0].is_nullable());
            assert!(fields[1].is_nullable());
            if let DataType::Struct(members) = fields[0].data_type() {
                assert!(members[0].is_nullable());
            }
        }
    }

    #[test]
    fn test_arrow_metadata_normalization_merges_ree_value_nullability() {
        let run_ends = Arc::new(Field::new("run_ends", DataType::Int32, false));
        let inner = Arc::new(Field::new("inner", DataType::Int32, true));
        let nested = Arc::new(Field::new(
            "nested",
            DataType::RunEndEncoded(Arc::clone(&run_ends), inner),
            false,
        ));
        let field = Field::new("outer", DataType::RunEndEncoded(run_ends, nested), false);

        let normalized = normalize_metadata_field(&field);
        assert_eq!(normalized.data_type(), &DataType::Int32);
        assert!(normalized.is_nullable());
    }

    #[test]
    fn test_arrow_metadata_normalization_descends_dictionary_values() {
        let ree = DataType::RunEndEncoded(
            Arc::new(Field::new("run_ends", DataType::Int32, false)),
            Arc::new(Field::new("values", DataType::Int64, true)),
        );
        let field = Field::new(
            "dictionary_ree",
            DataType::Dictionary(Box::new(DataType::Int8), Box::new(ree)),
            false,
        );

        assert!(needs_metadata_normalization(field.data_type()));
        let normalized = normalize_metadata_field(&field);
        assert_eq!(
            normalized.data_type(),
            &DataType::Dictionary(Box::new(DataType::Int8), Box::new(DataType::Int64))
        );
        assert!(normalized.is_nullable());
        assert!(!needs_metadata_normalization(normalized.data_type()));

        let mut props = WriterProperties::builder().build();
        add_encoded_arrow_schema_to_metadata(&Schema::new(vec![field]), &mut props);
        let encoded = props
            .key_value_metadata()
            .unwrap()
            .iter()
            .find(|kv| kv.key == crate::arrow::ARROW_SCHEMA_META_KEY)
            .unwrap()
            .value
            .as_deref()
            .unwrap();
        let decoded = get_arrow_schema_from_metadata(encoded).unwrap();
        assert_eq!(decoded.field(0), &normalized);
    }

    #[test]
    fn test_arrow_metadata_normalization_preserves_view_dictionary_siblings() {
        let dictionary = |dt| DataType::Dictionary(Box::new(DataType::Int8), Box::new(dt));
        let ree = DataType::RunEndEncoded(
            Arc::new(Field::new(
                Field::REE_RUN_ENDS_FIELD_DEFAULT_NAME,
                DataType::Int32,
                false,
            )),
            Arc::new(Field::new(
                Field::REE_VALUES_FIELD_DEFAULT_NAME,
                DataType::Int32,
                true,
            )),
        );
        let fields = vec![
            // This field forces normalization of the schema's metadata.
            Field::new(
                "nested",
                dictionary(DataType::Struct(
                    vec![Field::new("child", ree, false)].into(),
                )),
                false,
            ),
            // Scalar view dictionaries are reader-supported, just like Utf8
            // dictionaries. Normalizing their sibling must preserve these hints.
            Field::new("view", dictionary(DataType::Utf8View), true),
            Field::new("ordinary", dictionary(DataType::Utf8), true),
            Field::new("binary_view", dictionary(DataType::BinaryView), true),
        ];
        let schema = Schema::new(fields);
        assert!(needs_metadata_normalization(schema.field(0).data_type()));
        let mut props = WriterProperties::builder().build();
        add_encoded_arrow_schema_to_metadata(&schema, &mut props);
        let encoded = props
            .key_value_metadata()
            .unwrap()
            .iter()
            .find(|kv| kv.key == crate::arrow::ARROW_SCHEMA_META_KEY)
            .unwrap()
            .value
            .as_deref()
            .unwrap();
        let decoded = get_arrow_schema_from_metadata(encoded).unwrap();
        let DataType::Struct(children) = decoded.field(0).data_type() else {
            panic!("nested dictionary not erased")
        };
        assert_eq!(children[0].name(), "child");
        assert_eq!(children[0].data_type(), &DataType::Int32);
        assert!(children[0].is_nullable());
        assert!(!decoded.field(0).is_nullable());
        assert_eq!(decoded.field(1), schema.field(1));
        assert_eq!(decoded.field(2), schema.field(2));
        assert_eq!(decoded.field(3), schema.field(3));
        assert!(!needs_metadata_normalization(schema.field(1).data_type()));
        assert!(!needs_metadata_normalization(schema.field(3).data_type()));
        let parquet = ArrowSchemaConverter::new().convert(&schema).unwrap();
        assert_eq!(parquet.column(0).path().parts(), &["nested", "child"]);
        assert_eq!(parquet.column(0).max_def_level(), 1);
    }

    #[test]
    fn test_metadata() {
        let message_type = "
        message test_schema {
            OPTIONAL BINARY  string (STRING);
        }
        ";
        let parquet_group_type = parse_message_type(message_type).unwrap();

        let key_value_metadata = vec![
            KeyValue::new("foo".to_owned(), Some("bar".to_owned())),
            KeyValue::new("baz".to_owned(), None),
        ];

        let mut expected_metadata: HashMap<String, String> = HashMap::new();
        expected_metadata.insert("foo".to_owned(), "bar".to_owned());

        let parquet_schema = SchemaDescriptor::new(Arc::new(parquet_group_type));
        let converted_arrow_schema =
            parquet_to_arrow_schema(&parquet_schema, Some(&key_value_metadata)).unwrap();

        assert_eq!(converted_arrow_schema.metadata(), &expected_metadata);
    }

    #[test]
    fn test_arrow_schema_roundtrip() -> Result<()> {
        let meta = |a: &[(&str, &str)]| -> HashMap<String, String> {
            a.iter()
                .map(|(a, b)| (a.to_string(), b.to_string()))
                .collect()
        };

        let schema = Schema::new_with_metadata(
            vec![
                Field::new("c1", DataType::Utf8, false)
                    .with_metadata(meta(&[("Key", "Foo"), (PARQUET_FIELD_ID_META_KEY, "2")])),
                Field::new("c2", DataType::Binary, false),
                Field::new("c3", DataType::FixedSizeBinary(3), false),
                Field::new("c4", DataType::Boolean, false),
                Field::new("c5", DataType::Date32, false),
                Field::new("c6", DataType::Date64, false),
                Field::new("c7", DataType::Time32(TimeUnit::Second), false),
                Field::new("c8", DataType::Time32(TimeUnit::Millisecond), false),
                Field::new("c13", DataType::Time64(TimeUnit::Microsecond), false),
                Field::new("c14", DataType::Time64(TimeUnit::Nanosecond), false),
                Field::new("c15", DataType::Timestamp(TimeUnit::Second, None), false),
                Field::new(
                    "c16",
                    DataType::Timestamp(TimeUnit::Millisecond, Some("UTC".into())),
                    false,
                ),
                Field::new(
                    "c17",
                    DataType::Timestamp(TimeUnit::Microsecond, Some("Africa/Johannesburg".into())),
                    false,
                ),
                Field::new(
                    "c18",
                    DataType::Timestamp(TimeUnit::Nanosecond, None),
                    false,
                ),
                Field::new("c19", DataType::Interval(IntervalUnit::DayTime), false),
                Field::new("c20", DataType::Interval(IntervalUnit::YearMonth), false),
                Field::new_list(
                    "c21",
                    Field::new_list_field(DataType::Boolean, true)
                        .with_metadata(meta(&[("Key", "Bar"), (PARQUET_FIELD_ID_META_KEY, "5")])),
                    false,
                )
                .with_metadata(meta(&[(PARQUET_FIELD_ID_META_KEY, "4")])),
                Field::new(
                    "c22",
                    DataType::FixedSizeList(
                        Arc::new(Field::new_list_field(DataType::Boolean, true)),
                        5,
                    ),
                    false,
                ),
                Field::new_list(
                    "c23",
                    Field::new_large_list(
                        "inner",
                        Field::new_list_field(
                            DataType::Struct(
                                vec![
                                    Field::new("a", DataType::Int16, true),
                                    Field::new("b", DataType::Float64, false),
                                    Field::new("c", DataType::Float32, false),
                                    Field::new("d", DataType::Float16, false),
                                ]
                                .into(),
                            ),
                            false,
                        ),
                        true,
                    ),
                    false,
                ),
                Field::new(
                    "c24",
                    DataType::Struct(Fields::from(vec![
                        Field::new("a", DataType::Utf8, false),
                        Field::new("b", DataType::UInt16, false),
                    ])),
                    false,
                ),
                Field::new("c25", DataType::Interval(IntervalUnit::YearMonth), true),
                Field::new("c26", DataType::Interval(IntervalUnit::DayTime), true),
                // Duration types not supported
                // Field::new("c27", DataType::Duration(TimeUnit::Second), false),
                // Field::new("c28", DataType::Duration(TimeUnit::Millisecond), false),
                // Field::new("c29", DataType::Duration(TimeUnit::Microsecond), false),
                // Field::new("c30", DataType::Duration(TimeUnit::Nanosecond), false),
                #[expect(deprecated)]
                Field::new_dict(
                    "c31",
                    DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Utf8)),
                    true,
                    123,
                    true,
                )
                .with_metadata(meta(&[(PARQUET_FIELD_ID_META_KEY, "6")])),
                Field::new("c32", DataType::LargeBinary, true),
                Field::new("c33", DataType::LargeUtf8, true),
                Field::new_large_list(
                    "c34",
                    Field::new_list(
                        "inner",
                        Field::new_list_field(
                            DataType::Struct(
                                vec![
                                    Field::new("a", DataType::Int16, true),
                                    Field::new("b", DataType::Float64, true),
                                ]
                                .into(),
                            ),
                            true,
                        ),
                        true,
                    ),
                    true,
                ),
                Field::new("c35", DataType::Null, true),
                Field::new("c36", DataType::Decimal128(2, 1), false),
                Field::new("c37", DataType::Decimal256(50, 20), false),
                Field::new("c38", DataType::Decimal128(18, 12), true),
                Field::new_map(
                    "c39",
                    "key_value",
                    Field::new("key", DataType::Utf8, false),
                    Field::new_list("value", Field::new("element", DataType::Utf8, true), true),
                    false, // fails to roundtrip keys_sorted
                    true,
                ),
                Field::new_map(
                    "c40",
                    "my_entries",
                    Field::new("my_key", DataType::Utf8, false)
                        .with_metadata(meta(&[(PARQUET_FIELD_ID_META_KEY, "8")])),
                    Field::new_list(
                        "my_value",
                        Field::new_list_field(DataType::Utf8, true)
                            .with_metadata(meta(&[(PARQUET_FIELD_ID_META_KEY, "10")])),
                        true,
                    )
                    .with_metadata(meta(&[(PARQUET_FIELD_ID_META_KEY, "9")])),
                    false, // fails to roundtrip keys_sorted
                    true,
                )
                .with_metadata(meta(&[(PARQUET_FIELD_ID_META_KEY, "7")])),
                Field::new_map(
                    "c41",
                    "my_entries",
                    Field::new("my_key", DataType::Utf8, false),
                    Field::new_list(
                        "my_value",
                        Field::new_list_field(DataType::Utf8, true)
                            .with_metadata(meta(&[(PARQUET_FIELD_ID_META_KEY, "11")])),
                        true,
                    ),
                    false, // fails to roundtrip keys_sorted
                    false,
                ),
                Field::new("c42", DataType::Decimal32(5, 2), false),
                Field::new("c43", DataType::Decimal64(18, 12), true),
            ],
            meta(&[("Key", "Value")]),
        );

        // write to an empty parquet file so that schema is serialized
        let file = tempfile::tempfile().unwrap();
        let writer =
            ArrowWriter::try_new(file.try_clone().unwrap(), Arc::new(schema.clone()), None)?;
        writer.close()?;

        // read file back
        let arrow_reader = ParquetRecordBatchReaderBuilder::try_new(file).unwrap();

        // Check arrow schema
        let read_schema = arrow_reader.schema();
        assert_eq!(&schema, read_schema.as_ref());

        // Walk schema finding field IDs
        let mut stack = Vec::with_capacity(10);
        let mut out = Vec::with_capacity(10);

        let root = arrow_reader.parquet_schema().root_schema_ptr();
        stack.push((root.name().to_string(), root));

        while let Some((p, t)) = stack.pop() {
            if t.is_group() {
                for f in t.get_fields() {
                    stack.push((format!("{p}.{}", f.name()), f.clone()))
                }
            }

            let info = t.get_basic_info();
            if info.has_id() {
                out.push(format!("{p} -> {}", info.id()))
            }
        }
        out.sort_unstable();
        let out: Vec<_> = out.iter().map(|x| x.as_str()).collect();

        assert_eq!(
            &out,
            &[
                "arrow_schema.c1 -> 2",
                "arrow_schema.c21 -> 4",
                "arrow_schema.c21.list.item -> 5",
                "arrow_schema.c31 -> 6",
                "arrow_schema.c40 -> 7",
                "arrow_schema.c40.my_entries.my_key -> 8",
                "arrow_schema.c40.my_entries.my_value -> 9",
                "arrow_schema.c40.my_entries.my_value.list.item -> 10",
                "arrow_schema.c41.my_entries.my_value.list.item -> 11",
            ]
        );

        Ok(())
    }

    #[test]
    fn test_read_parquet_field_ids_raw() -> Result<()> {
        let meta = |a: &[(&str, &str)]| -> HashMap<String, String> {
            a.iter()
                .map(|(a, b)| (a.to_string(), b.to_string()))
                .collect()
        };
        let schema = Schema::new_with_metadata(
            vec![
                Field::new("c1", DataType::Utf8, true)
                    .with_metadata(meta(&[(PARQUET_FIELD_ID_META_KEY, "1")])),
                Field::new("c2", DataType::Utf8, true)
                    .with_metadata(meta(&[(PARQUET_FIELD_ID_META_KEY, "2")])),
            ],
            HashMap::new(),
        );

        let writer = ArrowWriter::try_new(vec![], Arc::new(schema.clone()), None)?;
        let parquet_bytes = writer.into_inner()?;

        let reader =
            crate::file::reader::SerializedFileReader::new(bytes::Bytes::from(parquet_bytes))?;
        let schema_descriptor = reader.metadata().file_metadata().schema_descr_ptr();

        // don't pass metadata so field ids are read from Parquet and not from serialized Arrow schema
        let arrow_schema = crate::arrow::parquet_to_arrow_schema(&schema_descriptor, None)?;

        let parq_schema_descr = ArrowSchemaConverter::new()
            .with_coerce_types(true)
            .convert(&arrow_schema)?;
        let parq_fields = parq_schema_descr.root_schema().get_fields();
        assert_eq!(parq_fields.len(), 2);
        assert_eq!(parq_fields[0].get_basic_info().id(), 1);
        assert_eq!(parq_fields[1].get_basic_info().id(), 2);

        Ok(())
    }

    #[test]
    fn test_arrow_schema_roundtrip_lists() -> Result<()> {
        let metadata = HashMap::from([("Key".to_string(), "Value".to_string())]);

        let schema = Schema::new_with_metadata(
            vec![
                Field::new_list("c21", Field::new("array", DataType::Boolean, true), false),
                Field::new(
                    "c22",
                    DataType::FixedSizeList(
                        Arc::new(Field::new("items", DataType::Boolean, false)),
                        5,
                    ),
                    false,
                ),
                Field::new_list(
                    "c23",
                    Field::new_large_list(
                        "items",
                        Field::new_struct(
                            "items",
                            vec![
                                Field::new("a", DataType::Int16, true),
                                Field::new("b", DataType::Float64, false),
                            ],
                            true,
                        ),
                        true,
                    ),
                    true,
                ),
            ],
            metadata,
        );

        // write to an empty parquet file so that schema is serialized
        let file = tempfile::tempfile().unwrap();
        let writer =
            ArrowWriter::try_new(file.try_clone().unwrap(), Arc::new(schema.clone()), None)?;
        writer.close()?;

        // read file back
        let arrow_reader = ParquetRecordBatchReaderBuilder::try_new(file).unwrap();
        let read_schema = arrow_reader.schema();
        assert_eq!(&schema, read_schema.as_ref());
        Ok(())
    }

    #[test]
    fn test_get_arrow_schema_from_metadata() {
        assert!(get_arrow_schema_from_metadata("").is_err());
    }

    #[test]
    #[cfg(feature = "arrow_canonical_extension_types")]
    fn arrow_uuid_to_parquet_uuid() -> Result<()> {
        use arrow_schema::extension::Uuid;
        let arrow_schema = Schema::new(vec![
            Field::new("uuid", DataType::FixedSizeBinary(16), false).with_extension_type(Uuid),
        ]);

        let parquet_schema = ArrowSchemaConverter::new().convert(&arrow_schema)?;

        assert_eq!(
            parquet_schema.column(0).logical_type_ref(),
            Some(&LogicalType::Uuid)
        );

        let arrow_schema = parquet_to_arrow_schema(&parquet_schema, None)?;
        assert_eq!(arrow_schema.field(0).try_extension_type::<Uuid>()?, Uuid);

        Ok(())
    }

    #[test]
    #[cfg(feature = "arrow_canonical_extension_types")]
    fn arrow_json_to_parquet_json() -> Result<()> {
        use arrow_schema::extension::Json;
        let arrow_schema = Schema::new(vec![
            Field::new("json", DataType::Utf8, false).with_extension_type(Json::default()),
        ]);

        let parquet_schema = ArrowSchemaConverter::new().convert(&arrow_schema)?;

        assert_eq!(
            parquet_schema.column(0).logical_type_ref(),
            Some(&LogicalType::Json)
        );

        let arrow_schema = parquet_to_arrow_schema(&parquet_schema, None)?;
        assert_eq!(
            arrow_schema.field(0).try_extension_type::<Json>()?,
            Json::default()
        );

        Ok(())
    }

    #[test]
    #[cfg(feature = "variant_experimental")]
    fn arrow_variant_to_parquet_variant() -> Result<()> {
        use parquet_variant_compute::VariantType;
        let variant_fields = Fields::from(vec![
            Field::new("metadata", DataType::BinaryView, false),
            Field::new("value", DataType::BinaryView, true),
        ]);
        let arrow_schema = Schema::new(vec![
            Field::new("v", DataType::Struct(variant_fields), true)
                .with_extension_type(VariantType),
        ]);

        let parquet_schema = ArrowSchemaConverter::new().convert(&arrow_schema)?;

        // The Parquet spec requires the VARIANT annotation to carry the
        // specification version (`VARIANT(1)`); Spark rejects files without it
        let root = parquet_schema.root_schema();
        assert_eq!(
            root.get_fields()[0].get_basic_info().logical_type_ref(),
            Some(&LogicalType::variant(Some(1)))
        );

        let arrow_schema = parquet_to_arrow_schema(&parquet_schema, None)?;
        assert!(
            arrow_schema
                .field(0)
                .has_valid_extension_type::<VariantType>()
        );

        Ok(())
    }

    #[test]
    fn test_parquet_to_arrow_field_levels_with_virtual_rejects_non_virtual() {
        let message_type = "
        message test_schema {
            REQUIRED INT32 id;
        }
        ";
        let parquet_schema = Arc::new(parse_message_type(message_type).unwrap());
        let descriptor = SchemaDescriptor::new(parquet_schema);

        // Try to pass a regular field (not a virtual column)
        let regular_field = Arc::new(Field::new("regular_column", DataType::Int64, false));
        let result = parquet_to_arrow_field_levels_with_virtual(
            &descriptor,
            ProjectionMask::all(),
            None,
            &[regular_field],
        );

        assert!(result.is_err());
        assert!(
            result
                .unwrap_err()
                .to_string()
                .contains("is not a virtual column")
        );
    }
}