parquet/

basic.rs

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

//! Contains Rust mappings for Thrift definition. This module contains only mappings for thrift
//! enums and unions. Thrift structs are handled elsewhere.
//! Refer to [`parquet.thrift`](https://github.com/apache/parquet-format/blob/master/src/main/thrift/parquet.thrift)
//! file to see raw definitions.

use std::io::Write;
use std::str::FromStr;
use std::{fmt, str};

pub use crate::compression::{BrotliLevel, GzipLevel, ZstdLevel};
use crate::file::metadata::HeapSize;
use crate::parquet_thrift::{
    ElementType, FieldType, ReadThrift, ThriftCompactInputProtocol, ThriftCompactOutputProtocol,
    WriteThrift, WriteThriftField, validate_list_type,
};
use crate::{
    thrift_enum, thrift_struct, thrift_union_all_empty, thrift_union_with_unknown,
    write_thrift_field,
};

use crate::errors::{ParquetError, Result};

// ----------------------------------------------------------------------
// Types from the Thrift definition

// ----------------------------------------------------------------------
// Mirrors thrift enum `Type`

thrift_enum!(
/// Types supported by Parquet.
///
/// These physical types are intended to be used in combination with the encodings to
/// control the on disk storage format.
/// For example INT16 is not included as a type since a good encoding of INT32
/// would handle this.
enum Type {
  BOOLEAN = 0;
  INT32 = 1;
  INT64 = 2;
  INT96 = 3;  // deprecated, only used by legacy implementations.
  FLOAT = 4;
  DOUBLE = 5;
  BYTE_ARRAY = 6;
  FIXED_LEN_BYTE_ARRAY = 7;
}
);

// ----------------------------------------------------------------------
// Mirrors thrift enum `ConvertedType`

// TODO(ets): Adding the `NONE` variant to this enum is a bit awkward. We should
// look into removing it and using `Option<ConvertedType>` instead.
thrift_enum!(
/// Common types (converted types) used by frameworks when using Parquet.
///
/// This helps map between types in those frameworks to the base types in Parquet.
/// This is only metadata and not needed to read or write the data.
///
/// This struct was renamed from `LogicalType` in version 4.0.0.
/// If targeting Parquet format 2.4.0 or above, please use [LogicalType] instead.
enum ConvertedType {
  /// Not defined in the spec, used internally to indicate no type conversion
  NONE = -1;

  /// A BYTE_ARRAY actually contains UTF8 encoded chars.
  UTF8 = 0;

  /// A map is converted as an optional field containing a repeated key/value pair.
  MAP = 1;

  /// A key/value pair is converted into a group of two fields.
  MAP_KEY_VALUE = 2;

  /// A list is converted into an optional field containing a repeated field for its
  /// values.
  LIST = 3;

  /// An enum is converted into a BYTE_ARRAY field
  ENUM = 4;

  /// A decimal value.
  ///
  /// This may be used to annotate BYTE_ARRAY or FIXED_LEN_BYTE_ARRAY primitive
  /// types. The underlying byte array stores the unscaled value encoded as two's
  /// complement using big-endian byte order (the most significant byte is the
  /// zeroth element). The value of the decimal is the value * 10^{-scale}.
  ///
  /// This must be accompanied by a (maximum) precision and a scale in the
  /// SchemaElement. The precision specifies the number of digits in the decimal
  /// and the scale stores the location of the decimal point. For example 1.23
  /// would have precision 3 (3 total digits) and scale 2 (the decimal point is
  /// 2 digits over).
  DECIMAL = 5;

  /// A date stored as days since Unix epoch, encoded as the INT32 physical type.
  DATE = 6;

  /// The total number of milliseconds since midnight. The value is stored as an INT32
  /// physical type.
  TIME_MILLIS = 7;

  /// The total number of microseconds since midnight. The value is stored as an INT64
  /// physical type.
  TIME_MICROS = 8;

  /// Date and time recorded as milliseconds since the Unix epoch.
  /// Recorded as a physical type of INT64.
  TIMESTAMP_MILLIS = 9;

  /// Date and time recorded as microseconds since the Unix epoch.
  /// The value is stored as an INT64 physical type.
  TIMESTAMP_MICROS = 10;

  /// An unsigned 8 bit integer value stored as INT32 physical type.
  UINT_8 = 11;

  /// An unsigned 16 bit integer value stored as INT32 physical type.
  UINT_16 = 12;

  /// An unsigned 32 bit integer value stored as INT32 physical type.
  UINT_32 = 13;

  /// An unsigned 64 bit integer value stored as INT64 physical type.
  UINT_64 = 14;

  /// A signed 8 bit integer value stored as INT32 physical type.
  INT_8 = 15;

  /// A signed 16 bit integer value stored as INT32 physical type.
  INT_16 = 16;

  /// A signed 32 bit integer value stored as INT32 physical type.
  INT_32 = 17;

  /// A signed 64 bit integer value stored as INT64 physical type.
  INT_64 = 18;

  /// A JSON document embedded within a single UTF8 column.
  JSON = 19;

   /// A BSON document embedded within a single BINARY column.
  BSON = 20;

  /// An interval of time
  ///
  /// This type annotates data stored as a FIXED_LEN_BYTE_ARRAY of length 12.
  /// This data is composed of three separate little endian unsigned integers.
  /// Each stores a component of a duration of time. The first integer identifies
  /// the number of months associated with the duration, the second identifies
  /// the number of days associated with the duration and the third identifies
  /// the number of milliseconds associated with the provided duration.
  /// This duration of time is independent of any particular timezone or date.
  INTERVAL = 21;
}
);

// ----------------------------------------------------------------------
// Mirrors thrift union `TimeUnit`

thrift_union_all_empty!(
/// Time unit for `Time` and `Timestamp` logical types.
union TimeUnit {
  1: MilliSeconds MILLIS
  2: MicroSeconds MICROS
  3: NanoSeconds NANOS
}
);

// ----------------------------------------------------------------------
// Mirrors thrift union `LogicalType`

thrift_struct!(
pub struct DecimalType {
  /// The number of digits in the decimal.
  1: required i32 scale
  /// The location of the decimal point.
  2: required i32 precision
}
);

thrift_struct!(
pub struct TimestampType {
  /// Whether the timestamp is adjusted to UTC.
  1: required bool is_adjusted_to_u_t_c
  /// The unit of time.
  2: required TimeUnit unit
}
);

/// Identical to [`TimestampType`]
pub use TimestampType as TimeType;

thrift_struct!(
pub struct IntType {
  /// The number of bits in the integer.
  1: required i8 bit_width
  /// Whether the integer is signed.
  2: required bool is_signed
}
);

thrift_struct!(
pub struct VariantType {
  /// The version of the variant specification that the variant was
  /// written with.
  1: optional i8 specification_version
}
);

thrift_struct!(
pub struct GeometryType {
  /// A custom CRS. If unset the CRS `OGC:CRS84` should be used, which means that the geometries
  /// must be stored in longitude, latitude based on the WGS84 datum.
  1: optional string crs;
}
);

thrift_struct!(
pub struct GeographyType {
  /// A custom CRS. If unset the CRS `OGC:CRS84` should be used.
  1: optional string crs;
  /// An optional algorithm can be set to correctly interpret edges interpolation
  /// of the geometries. If unset, the `SPHERICAL` algorithm should be used.
  2: optional EdgeInterpolationAlgorithm algorithm;
}
);

impl GeographyType {
    /// Accessor for the `GeographyType::algorithm` field. If this field is not set, this
    /// function returns the default value (currently [`EdgeInterpolationAlgorithm::SPHERICAL`]
    /// per the Parquet [specification]).
    ///
    /// [specification]: https://github.com/apache/parquet-format/blob/master/LogicalTypes.md#geography
    pub fn algorithm(&self) -> Option<EdgeInterpolationAlgorithm> {
        self.algorithm.or(Some(Default::default()))
    }
}

thrift_union_with_unknown!(
/// Logical types used by version 2.4.0+ of the Parquet format.
///
/// This is an *entirely new* struct as of version
/// 4.0.0. The struct previously named `LogicalType` was renamed to
/// [`ConvertedType`]. Please see the README.md for more details.
union LogicalType {
   /// A UTF8 encoded string.
   1:  String
   /// A map of key-value pairs.
   2:  Map
   /// A list of elements.
   3:  List
   /// A set of predefined values.
   4:  Enum
   /// A decimal value with a specified scale and precision.
   5:  (DecimalType) Decimal
   /// A date stored as days since Unix epoch.
   6:  Date
   /// A time stored as [`TimeUnit`] since midnight.
   7:  (TimeType) Time
   /// A timestamp stored as [`TimeUnit`] since Unix epoch.
   8:  (TimestampType) Timestamp
   // 9: reserved for INTERVAL
   /// An integer with a specified bit width and signedness.
   10: (IntType) Integer
   /// An unknown logical type.
   11: Unknown
   /// A JSON document.
   12: Json
   /// A BSON document.
   13: Bson
   /// A UUID.
   14: Uuid
   /// A 16-bit floating point number.
   15: Float16
   /// A Variant value.
   16: (VariantType) Variant
   /// A geospatial feature in the Well-Known Binary (WKB) format with linear/planar edges interpolation.
   17: (GeometryType) Geometry
   /// A geospatial feature in the WKB format with an explicit (non-linear/non-planar) edges interpolation.
   18: (GeographyType) Geography
}
);

impl LogicalType {
    /// Create a [`LogicalType::Integer`] variant with the given `bit_width` and `is_signed`
    pub fn integer(bit_width: i8, is_signed: bool) -> Self {
        Self::Integer(IntType {
            bit_width,
            is_signed,
        })
    }

    /// Create a [`LogicalType::Decimal`] variant with the given `scale` and `precision`
    pub fn decimal(scale: i32, precision: i32) -> Self {
        Self::Decimal(DecimalType { scale, precision })
    }

    /// Create a [`LogicalType::Time`] variant with the given `is_adjusted_to_u_t_c` and `unit`
    pub fn time(is_adjusted_to_u_t_c: bool, unit: TimeUnit) -> Self {
        Self::Time(TimeType {
            is_adjusted_to_u_t_c,
            unit,
        })
    }

    /// Create a [`LogicalType::Timestamp`] variant with the given `is_adjusted_to_u_t_c` and `unit`
    pub fn timestamp(is_adjusted_to_u_t_c: bool, unit: TimeUnit) -> Self {
        Self::Timestamp(TimestampType {
            is_adjusted_to_u_t_c,
            unit,
        })
    }

    /// Create a [`LogicalType::Variant`] variant with the given `specification_version`
    pub fn variant(specification_version: Option<i8>) -> Self {
        Self::Variant(VariantType {
            specification_version,
        })
    }

    /// Create a [`LogicalType::Geometry`] variant with the given `crs`
    pub fn geometry(crs: Option<String>) -> Self {
        Self::Geometry(GeometryType { crs })
    }

    /// Create a [`LogicalType::Geography`] variant with the given `crs` and `algorithm`
    pub fn geography(crs: Option<String>, algorithm: Option<EdgeInterpolationAlgorithm>) -> Self {
        Self::Geography(GeographyType { crs, algorithm })
    }
}

// ----------------------------------------------------------------------
// Mirrors thrift enum `FieldRepetitionType`
//

thrift_enum!(
/// Representation of field types in schema.
enum FieldRepetitionType {
  /// This field is required (can not be null) and each row has exactly 1 value.
  REQUIRED = 0;
  /// The field is optional (can be null) and each row has 0 or 1 values.
  OPTIONAL = 1;
  /// The field is repeated and can contain 0 or more values.
  REPEATED = 2;
}
);

/// Type alias for thrift `FieldRepetitionType`
pub type Repetition = FieldRepetitionType;

// ----------------------------------------------------------------------
// Mirrors thrift enum `Encoding`

thrift_enum!(
/// Encodings supported by Parquet.
///
/// Not all encodings are valid for all types. These enums are also used to specify the
/// encoding of definition and repetition levels.
///
/// By default this crate uses [Encoding::PLAIN], [Encoding::RLE], and [Encoding::RLE_DICTIONARY].
/// These provide very good encode and decode performance, whilst yielding reasonable storage
/// efficiency and being supported by all major parquet readers.
///
/// The delta encodings are also supported and will be used if a newer [WriterVersion] is
/// configured, however, it should be noted that these sacrifice encode and decode performance for
/// improved storage efficiency. This performance regression is particularly pronounced in the case
/// of record skipping as occurs during predicate push-down. It is recommended users assess the
/// performance impact when evaluating these encodings.
///
/// [WriterVersion]: crate::file::properties::WriterVersion
enum Encoding {
  /// Default encoding.
  /// - BOOLEAN - 1 bit per value. 0 is false; 1 is true.
  /// - INT32 - 4 bytes per value.  Stored as little-endian.
  /// - INT64 - 8 bytes per value.  Stored as little-endian.
  /// - FLOAT - 4 bytes per value.  IEEE. Stored as little-endian.
  /// - DOUBLE - 8 bytes per value.  IEEE. Stored as little-endian.
  /// - BYTE_ARRAY - 4 byte length stored as little endian, followed by bytes.
  /// - FIXED_LEN_BYTE_ARRAY - Just the bytes.
  PLAIN = 0;
  //  GROUP_VAR_INT = 1;
  /// **Deprecated** dictionary encoding.
  ///
  /// The values in the dictionary are encoded using PLAIN encoding.
  /// Since it is deprecated, RLE_DICTIONARY encoding is used for a data page, and
  /// PLAIN encoding is used for dictionary page.
  PLAIN_DICTIONARY = 2;
  /// Group packed run length encoding.
  ///
  /// Usable for definition/repetition levels encoding and boolean values.
  RLE = 3;
  /// **Deprecated** Bit-packed encoding.
  ///
  /// This can only be used if the data has a known max width.
  /// Usable for definition/repetition levels encoding.
  ///
  /// There are compatibility issues with files using this encoding.
  /// The parquet standard specifies the bits to be packed starting from the
  /// most-significant bit, several implementations do not follow this bit order.
  /// Several other implementations also have issues reading this encoding
  /// because of incorrect assumptions about the length of the encoded data.
  ///
  /// The RLE/bit-packing hybrid is more cpu and memory efficient and should be used instead.
  #[deprecated(
      since = "51.0.0",
      note = "Please see documentation for compatibility issues and use the RLE/bit-packing hybrid encoding instead"
  )]
  BIT_PACKED = 4;
  /// Delta encoding for integers, either INT32 or INT64.
  ///
  /// Works best on sorted data.
  DELTA_BINARY_PACKED = 5;
  /// Encoding for byte arrays to separate the length values and the data.
  ///
  /// The lengths are encoded using DELTA_BINARY_PACKED encoding.
  DELTA_LENGTH_BYTE_ARRAY = 6;
  /// Incremental encoding for byte arrays.
  ///
  /// Prefix lengths are encoded using DELTA_BINARY_PACKED encoding.
  /// Suffixes are stored using DELTA_LENGTH_BYTE_ARRAY encoding.
  DELTA_BYTE_ARRAY = 7;
  /// Dictionary encoding.
  ///
  /// The ids are encoded using the RLE encoding.
  RLE_DICTIONARY = 8;
  /// Encoding for fixed-width data.
  ///
  /// K byte-streams are created where K is the size in bytes of the data type.
  /// The individual bytes of a value are scattered to the corresponding stream and
  /// the streams are concatenated.
  /// This itself does not reduce the size of the data but can lead to better compression
  /// afterwards. Note that the use of this encoding with FIXED_LEN_BYTE_ARRAY(N) data may
  /// perform poorly for large values of N.
  BYTE_STREAM_SPLIT = 9;
}
);

impl FromStr for Encoding {
    type Err = ParquetError;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        match s {
            "PLAIN" | "plain" => Ok(Encoding::PLAIN),
            "PLAIN_DICTIONARY" | "plain_dictionary" => Ok(Encoding::PLAIN_DICTIONARY),
            "RLE" | "rle" => Ok(Encoding::RLE),
            #[allow(deprecated)]
            "BIT_PACKED" | "bit_packed" => Ok(Encoding::BIT_PACKED),
            "DELTA_BINARY_PACKED" | "delta_binary_packed" => Ok(Encoding::DELTA_BINARY_PACKED),
            "DELTA_LENGTH_BYTE_ARRAY" | "delta_length_byte_array" => {
                Ok(Encoding::DELTA_LENGTH_BYTE_ARRAY)
            }
            "DELTA_BYTE_ARRAY" | "delta_byte_array" => Ok(Encoding::DELTA_BYTE_ARRAY),
            "RLE_DICTIONARY" | "rle_dictionary" => Ok(Encoding::RLE_DICTIONARY),
            "BYTE_STREAM_SPLIT" | "byte_stream_split" => Ok(Encoding::BYTE_STREAM_SPLIT),
            _ => Err(general_err!("unknown encoding: {}", s)),
        }
    }
}

/// A bitmask representing the [`Encoding`]s employed while encoding a Parquet column chunk.
///
/// The Parquet [`ColumnMetaData`] struct contains an array that indicates what encodings were
/// used when writing that column chunk. For memory and performance reasons, this crate reduces
/// that array to bitmask, where each bit position represents a different [`Encoding`]. This
/// struct contains that bitmask, and provides methods to interact with the data.
///
/// # Example
/// ```no_run
/// # use parquet::file::metadata::ParquetMetaDataReader;
/// # use parquet::basic::Encoding;
/// # fn open_parquet_file(path: &str) -> std::fs::File { unimplemented!(); }
/// // read parquet metadata from a file
/// let file = open_parquet_file("some_path.parquet");
/// let mut reader = ParquetMetaDataReader::new();
/// reader.try_parse(&file).unwrap();
/// let metadata = reader.finish().unwrap();
///
/// // find the encodings used by the first column chunk in the first row group
/// let col_meta = metadata.row_group(0).column(0);
/// let encodings = col_meta.encodings_mask();
///
/// // check to see if a particular encoding was used
/// let used_rle = encodings.is_set(Encoding::RLE);
///
/// // check to see if all of a set of encodings were used
/// let used_all = encodings.all_set([Encoding::RLE, Encoding::PLAIN].iter());
///
/// // convert mask to a Vec<Encoding>
/// let encodings_vec = encodings.encodings().collect::<Vec<_>>();
/// ```
///
/// [`ColumnMetaData`]: https://github.com/apache/parquet-format/blob/9fd57b59e0ce1a82a69237dcf8977d3e72a2965d/src/main/thrift/parquet.thrift#L875
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct EncodingMask(i32);

impl EncodingMask {
    /// Highest valued discriminant in the [`Encoding`] enum
    const MAX_ENCODING: i32 = Encoding::MAX_DISCRIMINANT;
    /// A mask consisting of unused bit positions, used for validation. This includes the never
    /// used GROUP_VAR_INT encoding value of `1`.
    const ALLOWED_MASK: u32 =
        !(1u32 << (EncodingMask::MAX_ENCODING as u32 + 1)).wrapping_sub(1) | (1 << 1);

    /// Attempt to create a new `EncodingMask` from an integer.
    ///
    /// This will return an error if a bit outside the allowable range is set.
    pub fn try_new(val: i32) -> Result<Self> {
        if val as u32 & Self::ALLOWED_MASK != 0 {
            return Err(general_err!("Attempt to create invalid mask: 0x{:x}", val));
        }
        Ok(Self(val))
    }

    /// Return an integer representation of this `EncodingMask`.
    pub fn as_i32(&self) -> i32 {
        self.0
    }

    /// Create a new `EncodingMask` from a collection of [`Encoding`]s.
    pub fn new_from_encodings<'a>(encodings: impl Iterator<Item = &'a Encoding>) -> Self {
        let mut mask = 0;
        for &e in encodings {
            mask |= 1 << (e as i32);
        }
        Self(mask)
    }

    /// Mark the given [`Encoding`] as present in this mask.
    pub fn insert(&mut self, val: Encoding) {
        self.0 |= 1 << (val as i32);
    }

    /// Test if a given [`Encoding`] is present in this mask.
    pub fn is_set(&self, val: Encoding) -> bool {
        self.0 & (1 << (val as i32)) != 0
    }

    /// Test if this mask has only the bit for the given [`Encoding`] set.
    pub fn is_only(&self, val: Encoding) -> bool {
        self.0 == (1 << (val as i32))
    }

    /// Test if all [`Encoding`]s in a given set are present in this mask.
    pub fn all_set<'a>(&self, mut encodings: impl Iterator<Item = &'a Encoding>) -> bool {
        encodings.all(|&e| self.is_set(e))
    }

    /// Return an iterator over all [`Encoding`]s present in this mask.
    pub fn encodings(&self) -> impl Iterator<Item = Encoding> {
        Self::mask_to_encodings_iter(self.0)
    }

    fn mask_to_encodings_iter(mask: i32) -> impl Iterator<Item = Encoding> {
        (0..=Self::MAX_ENCODING)
            .filter(move |i| mask & (1 << i) != 0)
            .map(i32_to_encoding)
    }
}

impl HeapSize for EncodingMask {
    fn heap_size(&self) -> usize {
        0 // no heap allocations
    }
}

impl<'a, R: ThriftCompactInputProtocol<'a>> ReadThrift<'a, R> for EncodingMask {
    fn read_thrift(prot: &mut R) -> Result<Self> {
        let mut mask = 0;

        // This reads a Thrift `list<Encoding>` and turns it into a bitmask
        let list_ident = prot.read_list_begin()?;
        // check for enum (encoded as I32)
        validate_list_type(ElementType::I32, &list_ident)?;
        for _ in 0..list_ident.size {
            let val = Encoding::read_thrift(prot)?;
            mask |= 1 << val as i32;
        }
        Ok(Self(mask))
    }
}

#[allow(deprecated)]
fn i32_to_encoding(val: i32) -> Encoding {
    match val {
        0 => Encoding::PLAIN,
        2 => Encoding::PLAIN_DICTIONARY,
        3 => Encoding::RLE,
        4 => Encoding::BIT_PACKED,
        5 => Encoding::DELTA_BINARY_PACKED,
        6 => Encoding::DELTA_LENGTH_BYTE_ARRAY,
        7 => Encoding::DELTA_BYTE_ARRAY,
        8 => Encoding::RLE_DICTIONARY,
        9 => Encoding::BYTE_STREAM_SPLIT,
        _ => panic!("Impossible encoding {val}"),
    }
}

// ----------------------------------------------------------------------
// Mirrors thrift enum `CompressionCodec`

thrift_enum!(
/// Supported compression algorithms.
///
/// Codecs added in format version X.Y can be read by readers based on X.Y and later.
/// Codec support may vary between readers based on the format version and
/// libraries available at runtime.
///
/// See [Compression.md] for a detailed specification of these algorithms.
///
/// [Compression.md]: https://github.com/apache/parquet-format/blob/master/Compression.md
enum CompressionCodec {
  UNCOMPRESSED = 0;
  SNAPPY = 1;
  GZIP = 2;
  LZO = 3;
  BROTLI = 4;  // Added in 2.4
  LZ4 = 5;     // DEPRECATED (Added in 2.4)
  ZSTD = 6;    // Added in 2.4
  LZ4_RAW = 7; // Added in 2.9
}
);

// NOTE: This enum likely belongs in file::properties now, but moving it there would be a
// breaking API change, that's probably not worth the pain. If a new codec is added to the
// Parquet specification, or any other breaking changes are made to this enum, this can be
// revisited.

/// Supported block compression algorithms.
///
/// Block compression can yield non-trivial improvements to storage efficiency at the expense
/// of potentially significantly worse encode and decode performance. Many applications,
/// especially those making use of high-throughput and low-cost commodity object storage,
/// may find storage efficiency less important than decode throughput, and therefore may
/// wish to not make use of block compression.
///
/// The writers in this crate default to no block compression for this reason.
///
/// Applications that do still wish to use block compression, will find [`Compression::ZSTD`]
/// to provide a good balance of compression, performance, and ecosystem support. Alternatively,
/// [`Compression::LZ4_RAW`] provides much faster decompression speeds, at the cost of typically
/// worse compression ratios. However, it is not as widely supported by the ecosystem, with the
/// Hadoop ecosystem historically favoring the non-standard and now deprecated [`Compression::LZ4`].
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[allow(non_camel_case_types)]
pub enum Compression {
    /// No compression.
    UNCOMPRESSED,
    /// [Snappy compression](https://en.wikipedia.org/wiki/Snappy_(compression))
    SNAPPY,
    /// [Gzip compression](https://www.ietf.org/rfc/rfc1952.txt)
    GZIP(GzipLevel),
    /// [LZO compression](https://en.wikipedia.org/wiki/Lempel%E2%80%93Ziv%E2%80%93Oberhumer)
    LZO,
    /// [Brotli compression](https://datatracker.ietf.org/doc/html/rfc7932)
    BROTLI(BrotliLevel),
    /// [LZ4 compression](https://lz4.org/), [(deprecated)](https://issues.apache.org/jira/browse/PARQUET-2032)
    LZ4,
    /// [ZSTD compression](https://datatracker.ietf.org/doc/html/rfc8878)
    ZSTD(ZstdLevel),
    /// [LZ4 compression](https://lz4.org/).
    LZ4_RAW,
}

impl From<CompressionCodec> for Compression {
    fn from(value: CompressionCodec) -> Self {
        match value {
            CompressionCodec::UNCOMPRESSED => Compression::UNCOMPRESSED,
            CompressionCodec::SNAPPY => Compression::SNAPPY,
            CompressionCodec::GZIP => Compression::GZIP(Default::default()),
            CompressionCodec::LZO => Compression::LZO,
            CompressionCodec::BROTLI => Compression::BROTLI(Default::default()),
            CompressionCodec::LZ4 => Compression::LZ4,
            CompressionCodec::ZSTD => Compression::ZSTD(Default::default()),
            CompressionCodec::LZ4_RAW => Compression::LZ4_RAW,
        }
    }
}

impl From<Compression> for CompressionCodec {
    fn from(value: Compression) -> Self {
        match value {
            Compression::UNCOMPRESSED => CompressionCodec::UNCOMPRESSED,
            Compression::SNAPPY => CompressionCodec::SNAPPY,
            Compression::GZIP(_) => CompressionCodec::GZIP,
            Compression::LZO => CompressionCodec::LZO,
            Compression::BROTLI(_) => CompressionCodec::BROTLI,
            Compression::LZ4 => CompressionCodec::LZ4,
            Compression::ZSTD(_) => CompressionCodec::ZSTD,
            Compression::LZ4_RAW => CompressionCodec::LZ4_RAW,
        }
    }
}

fn split_compression_string(str_setting: &str) -> Result<(&str, Option<i32>), ParquetError> {
    let split_setting = str_setting.split_once('(');

    match split_setting {
        Some((codec, level_str)) => {
            let level = &level_str[..level_str.len() - 1]
                .parse::<i32>()
                .map_err(|_| {
                    ParquetError::General(format!("invalid compression level: {level_str}"))
                })?;
            Ok((codec, Some(*level)))
        }
        None => Ok((str_setting, None)),
    }
}

fn check_level_is_none(level: &Option<i32>) -> Result<(), ParquetError> {
    if level.is_some() {
        return Err(ParquetError::General(
            "compression level is not supported".to_string(),
        ));
    }

    Ok(())
}

fn require_level(codec: &str, level: Option<i32>) -> Result<i32, ParquetError> {
    level.ok_or(ParquetError::General(format!(
        "{codec} requires a compression level",
    )))
}

impl FromStr for Compression {
    type Err = ParquetError;

    fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
        let (codec, level) = split_compression_string(s)?;

        let c = match codec {
            "UNCOMPRESSED" | "uncompressed" => {
                check_level_is_none(&level)?;
                Compression::UNCOMPRESSED
            }
            "SNAPPY" | "snappy" => {
                check_level_is_none(&level)?;
                Compression::SNAPPY
            }
            "GZIP" | "gzip" => {
                let level = require_level(codec, level)?;
                Compression::GZIP(GzipLevel::try_new(level.try_into()?)?)
            }
            "LZO" | "lzo" => {
                check_level_is_none(&level)?;
                Compression::LZO
            }
            "BROTLI" | "brotli" => {
                let level = require_level(codec, level)?;
                Compression::BROTLI(BrotliLevel::try_new(level.try_into()?)?)
            }
            "LZ4" | "lz4" => {
                check_level_is_none(&level)?;
                Compression::LZ4
            }
            "ZSTD" | "zstd" => {
                let level = require_level(codec, level)?;
                Compression::ZSTD(ZstdLevel::try_new(level)?)
            }
            "LZ4_RAW" | "lz4_raw" => {
                check_level_is_none(&level)?;
                Compression::LZ4_RAW
            }
            _ => {
                return Err(ParquetError::General(format!(
                    "unsupport compression {codec}"
                )));
            }
        };

        Ok(c)
    }
}

// ----------------------------------------------------------------------
// Mirrors thrift enum `PageType`

thrift_enum!(
/// Available data pages for Parquet file format.
/// Note that some of the page types may not be supported.
enum PageType {
  DATA_PAGE = 0;
  INDEX_PAGE = 1;
  DICTIONARY_PAGE = 2;
  DATA_PAGE_V2 = 3;
}
);

// ----------------------------------------------------------------------
// Mirrors thrift enum `BoundaryOrder`

thrift_enum!(
/// Enum to annotate whether lists of min/max elements inside ColumnIndex
/// are ordered and if so, in which direction.
enum BoundaryOrder {
  UNORDERED = 0;
  ASCENDING = 1;
  DESCENDING = 2;
}
);

// ----------------------------------------------------------------------
// Mirrors thrift enum `EdgeInterpolationAlgorithm`

// this is hand coded to allow for the _Unknown variant (allows this to be forward compatible)

/// Edge interpolation algorithm for [`LogicalType::Geography`]
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
#[repr(i32)]
#[derive(Default)]
pub enum EdgeInterpolationAlgorithm {
    /// Edges are interpolated as geodesics on a sphere.
    #[default]
    SPHERICAL = 0,
    /// <https://en.wikipedia.org/wiki/Vincenty%27s_formulae>
    VINCENTY = 1,
    /// Thomas, Paul D. Spheroidal geodesics, reference systems, & local geometry. US Naval Oceanographic Office, 1970
    THOMAS = 2,
    /// Thomas, Paul D. Mathematical models for navigation systems. US Naval Oceanographic Office, 1965.
    ANDOYER = 3,
    /// Karney, Charles FF. "Algorithms for geodesics." Journal of Geodesy 87 (2013): 43-55
    KARNEY = 4,
    /// Unknown algorithm
    _Unknown(i32),
}

#[cfg(feature = "geospatial")]
impl EdgeInterpolationAlgorithm {
    /// Converts an [`EdgeInterpolationAlgorithm`] into its corresponding algorithm defined by
    /// [`parquet_geospatial::WkbEdges`].
    ///
    /// This method will only return an Err if the [`EdgeInterpolationAlgorithm`] is the `_Unknown`
    /// variant.
    pub fn try_as_edges(&self) -> Result<parquet_geospatial::WkbEdges> {
        match &self {
            Self::SPHERICAL => Ok(parquet_geospatial::WkbEdges::Spherical),
            Self::VINCENTY => Ok(parquet_geospatial::WkbEdges::Vincenty),
            Self::THOMAS => Ok(parquet_geospatial::WkbEdges::Thomas),
            Self::ANDOYER => Ok(parquet_geospatial::WkbEdges::Andoyer),
            Self::KARNEY => Ok(parquet_geospatial::WkbEdges::Karney),
            unknown => Err(general_err!(
                "Unknown edge interpolation algorithm: {}",
                unknown
            )),
        }
    }
}

impl fmt::Display for EdgeInterpolationAlgorithm {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        f.write_fmt(format_args!("{0:?}", self))
    }
}

#[cfg(feature = "geospatial")]
impl From<parquet_geospatial::WkbEdges> for EdgeInterpolationAlgorithm {
    fn from(value: parquet_geospatial::WkbEdges) -> Self {
        match value {
            parquet_geospatial::WkbEdges::Spherical => Self::SPHERICAL,
            parquet_geospatial::WkbEdges::Vincenty => Self::VINCENTY,
            parquet_geospatial::WkbEdges::Thomas => Self::THOMAS,
            parquet_geospatial::WkbEdges::Andoyer => Self::ANDOYER,
            parquet_geospatial::WkbEdges::Karney => Self::KARNEY,
        }
    }
}

impl FromStr for EdgeInterpolationAlgorithm {
    type Err = ParquetError;

    fn from_str(s: &str) -> Result<Self> {
        match s.to_ascii_uppercase().as_str() {
            "SPHERICAL" => Ok(EdgeInterpolationAlgorithm::SPHERICAL),
            "VINCENTY" => Ok(EdgeInterpolationAlgorithm::VINCENTY),
            "THOMAS" => Ok(EdgeInterpolationAlgorithm::THOMAS),
            "ANDOYER" => Ok(EdgeInterpolationAlgorithm::ANDOYER),
            "KARNEY" => Ok(EdgeInterpolationAlgorithm::KARNEY),
            unknown => Err(general_err!(
                "Unknown edge interpolation algorithm: {}",
                unknown
            )),
        }
    }
}

impl<'a, R: ThriftCompactInputProtocol<'a>> ReadThrift<'a, R> for EdgeInterpolationAlgorithm {
    fn read_thrift(prot: &mut R) -> Result<Self> {
        let val = prot.read_i32()?;
        match val {
            0 => Ok(Self::SPHERICAL),
            1 => Ok(Self::VINCENTY),
            2 => Ok(Self::THOMAS),
            3 => Ok(Self::ANDOYER),
            4 => Ok(Self::KARNEY),
            _ => Ok(Self::_Unknown(val)),
        }
    }
}

impl WriteThrift for EdgeInterpolationAlgorithm {
    const ELEMENT_TYPE: ElementType = ElementType::I32;
    fn write_thrift<W: Write>(&self, writer: &mut ThriftCompactOutputProtocol<W>) -> Result<()> {
        let val: i32 = match *self {
            Self::SPHERICAL => 0,
            Self::VINCENTY => 1,
            Self::THOMAS => 2,
            Self::ANDOYER => 3,
            Self::KARNEY => 4,
            Self::_Unknown(i) => i,
        };
        writer.write_i32(val)
    }
}

write_thrift_field!(EdgeInterpolationAlgorithm, FieldType::I32);

// ----------------------------------------------------------------------
// Mirrors thrift union `BloomFilterAlgorithm`

thrift_union_all_empty!(
/// The algorithm used in Bloom filter.
union BloomFilterAlgorithm {
  /// Block-based Bloom filter.
  1: SplitBlockAlgorithm BLOCK;
}
);

// ----------------------------------------------------------------------
// Mirrors thrift union `BloomFilterHash`

thrift_union_all_empty!(
/// The hash function used in Bloom filter. This function takes the hash of a column value
/// using plain encoding.
union BloomFilterHash {
  /// xxHash Strategy.
  1: XxHash XXHASH;
}
);

// ----------------------------------------------------------------------
// Mirrors thrift union `BloomFilterCompression`

thrift_union_all_empty!(
/// The compression used in the Bloom filter.
union BloomFilterCompression {
  1: Uncompressed UNCOMPRESSED;
}
);

// ----------------------------------------------------------------------
// Mirrors thrift union `ColumnOrder`

/// Sort order for page and column statistics.
///
/// Types are associated with sort orders and column stats are aggregated using a sort
/// order, and a sort order should be considered when comparing values with statistics
/// min/max.
///
/// See [`ColumnOrder`] for more information.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[allow(non_camel_case_types)]
pub enum SortOrder {
    /// Signed (either value or legacy byte-wise) comparison.
    SIGNED,
    /// Unsigned (depending on physical type either value or byte-wise) comparison.
    UNSIGNED,
    /// Comparison is undefined.
    UNDEFINED,
    /// Use IEEE 754 total order.
    TOTAL_ORDER,
    /// Use INT96 timestamp order (see [parquet-format/#584] and the [Thrift spec]).
    ///
    /// [parquet-format/#584]: https://github.com/apache/parquet-format/pull/584
    /// [Thrift spec]: https://github.com/apache/parquet-format/blob/2076361bb64e2de9ca6a8d06eda025a6fa4e9df6/src/main/thrift/parquet.thrift#L1230-L1233
    INT96_TIMESTAMP,
}

impl SortOrder {
    /// Returns true if this is [`Self::SIGNED`]
    pub fn is_signed(&self) -> bool {
        matches!(self, Self::SIGNED)
    }
}

/// Column order that specifies what method was used to aggregate min/max values for
/// statistics.
///
/// Prior to version 2.4.0, Parquet used signed comparisons when computing min and max
/// values for statistics. This caused problems for UTF8 encoded strings, so the
/// [`ColumnOrder`] union was added, initially with a single variant `TYPE_ORDER`. The
/// sort order for columns was then defined based on the logical or physical type of
/// the column, and could use either signed comparison, unsigned comparison, or for some
/// types be left undefined. Since then several new `ColumnOrder`s have been added to the
/// specification.
///
/// In this crate, the `ColumnOrder` found in the footer is represented by this enum. To
/// convey what actual sort order to use, this crate maps the `ColumnOrder` along with the
/// physical and logical type to a [`SortOrder`]. It is this [`SortOrder`] that is used
/// internally when deciding how to compute the min/max statistics.
///
/// If column order is undefined, then it is the legacy behaviour and all values should
/// be compared as signed values/bytes.
///
/// [`ColumnOrder`]: https://github.com/apache/parquet-format/blob/2076361bb64e2de9ca6a8d06eda025a6fa4e9df6/src/main/thrift/parquet.thrift#L1103
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[allow(non_camel_case_types)]
pub enum ColumnOrder {
    /// Column uses the order defined by its logical or physical type
    /// (if there is no logical type), parquet-format 2.4.0+.
    TYPE_DEFINED_ORDER(SortOrder),
    /// Column ordering to use for floating point types.
    IEEE_754_TOTAL_ORDER,
    /// Column ordering to use for INT96 types.
    INT96_TIMESTAMP_ORDER,
    // The following are not defined in the Parquet spec and should always be last.
    /// Undefined column order, means legacy behaviour before parquet-format 2.4.0.
    /// Sort order is always SIGNED.
    UNDEFINED,
    /// An unknown but present ColumnOrder. Statistics with an unknown `ColumnOrder`
    /// will be ignored.
    UNKNOWN,
}

impl ColumnOrder {
    /// Returns the `ColumnOrder` for a physical/logical type.
    pub fn column_order_for_type(
        logical_type: Option<&LogicalType>,
        converted_type: ConvertedType,
        physical_type: Type,
    ) -> ColumnOrder {
        if Some(&LogicalType::Float16) == logical_type
            || matches!(physical_type, Type::FLOAT | Type::DOUBLE)
        {
            ColumnOrder::IEEE_754_TOTAL_ORDER
        } else if matches!(physical_type, Type::INT96) {
            ColumnOrder::INT96_TIMESTAMP_ORDER
        } else {
            let sort_order =
                Self::get_sort_order_for_type(logical_type, converted_type, physical_type, true);
            ColumnOrder::TYPE_DEFINED_ORDER(sort_order)
        }
    }

    /// Returns sort order for a physical/logical type.
    ///
    /// `is_type_defined` indicates whether the column order for this type is
    /// [`ColumnOrder::TYPE_DEFINED_ORDER`].
    ///
    /// It is now preferred to obtain this via [`Self::sort_order`].
    #[deprecated(since = "60.0.0", note = "use `ColumnOrder::sort_order` instead")]
    pub fn sort_order_for_type(
        logical_type: Option<&LogicalType>,
        converted_type: ConvertedType,
        physical_type: Type,
        is_type_defined: bool,
    ) -> SortOrder {
        ColumnOrder::get_sort_order_for_type(
            logical_type,
            converted_type,
            physical_type,
            is_type_defined,
        )
    }

    // this is pub(crate) so it can be used in the thrift parser to correctly instantiate
    // the column_orders vec
    pub(crate) fn get_sort_order_for_type(
        logical_type: Option<&LogicalType>,
        converted_type: ConvertedType,
        physical_type: Type,
        is_type_defined: bool,
    ) -> SortOrder {
        match logical_type {
            Some(logical) => match logical {
                LogicalType::String | LogicalType::Enum | LogicalType::Json | LogicalType::Bson => {
                    SortOrder::UNSIGNED
                }
                LogicalType::Integer(int) => match int.is_signed {
                    true => SortOrder::SIGNED,
                    false => SortOrder::UNSIGNED,
                },
                LogicalType::Map | LogicalType::List => SortOrder::UNDEFINED,
                LogicalType::Decimal(_) => SortOrder::SIGNED,
                LogicalType::Date => SortOrder::SIGNED,
                LogicalType::Time(_) => SortOrder::SIGNED,
                LogicalType::Timestamp(_) => SortOrder::SIGNED,
                LogicalType::Unknown => SortOrder::UNDEFINED,
                LogicalType::Uuid => SortOrder::UNSIGNED,
                LogicalType::Float16 => {
                    if is_type_defined {
                        SortOrder::SIGNED
                    } else {
                        SortOrder::TOTAL_ORDER
                    }
                }
                LogicalType::Variant(_)
                | LogicalType::Geometry(_)
                | LogicalType::Geography(_)
                | LogicalType::_Unknown { .. } => SortOrder::UNDEFINED,
            },
            // Fall back to converted type
            None => Self::get_converted_sort_order(converted_type, physical_type, is_type_defined),
        }
    }

    fn get_converted_sort_order(
        converted_type: ConvertedType,
        physical_type: Type,
        is_type_defined: bool,
    ) -> SortOrder {
        match converted_type {
            // Unsigned byte-wise comparison.
            ConvertedType::UTF8
            | ConvertedType::JSON
            | ConvertedType::BSON
            | ConvertedType::ENUM => SortOrder::UNSIGNED,

            ConvertedType::INT_8
            | ConvertedType::INT_16
            | ConvertedType::INT_32
            | ConvertedType::INT_64 => SortOrder::SIGNED,

            ConvertedType::UINT_8
            | ConvertedType::UINT_16
            | ConvertedType::UINT_32
            | ConvertedType::UINT_64 => SortOrder::UNSIGNED,

            // Signed comparison of the represented value.
            ConvertedType::DECIMAL => SortOrder::SIGNED,

            ConvertedType::DATE => SortOrder::SIGNED,

            ConvertedType::TIME_MILLIS
            | ConvertedType::TIME_MICROS
            | ConvertedType::TIMESTAMP_MILLIS
            | ConvertedType::TIMESTAMP_MICROS => SortOrder::SIGNED,

            ConvertedType::INTERVAL => SortOrder::UNDEFINED,

            ConvertedType::LIST | ConvertedType::MAP | ConvertedType::MAP_KEY_VALUE => {
                SortOrder::UNDEFINED
            }

            // Fall back to physical type.
            ConvertedType::NONE => Self::get_default_sort_order(physical_type, is_type_defined),
        }
    }

    /// Returns default sort order based on physical type.
    fn get_default_sort_order(physical_type: Type, is_type_defined: bool) -> SortOrder {
        match physical_type {
            // Order: false, true
            Type::BOOLEAN => SortOrder::UNSIGNED,
            Type::INT32 | Type::INT64 => SortOrder::SIGNED,
            Type::INT96 => {
                if is_type_defined {
                    SortOrder::UNDEFINED
                } else {
                    SortOrder::INT96_TIMESTAMP
                }
            }
            // Notes to remember when comparing float/double values:
            // If legacy TYPE_DEFINED_ORDER is specified:
            //   If the min is a NaN, it should be ignored.
            //   If the max is a NaN, it should be ignored.
            //   If the min is +0, the row group may contain -0 values as well.
            //   If the max is -0, the row group may contain +0 values as well.
            //   When looking for NaN values, min and max should be ignored.
            // If IEEE_754_TOTAL_ORDER:
            //   Examine nan_count to see if NaNs are present.
            //   If min/max are NaN, that means only NaNs are present.
            //   If min/max are not NaN, they are ordered according to total order.
            Type::FLOAT | Type::DOUBLE => {
                if is_type_defined {
                    SortOrder::SIGNED
                } else {
                    SortOrder::TOTAL_ORDER
                }
            }
            // Unsigned byte-wise comparison
            Type::BYTE_ARRAY | Type::FIXED_LEN_BYTE_ARRAY => SortOrder::UNSIGNED,
        }
    }

    /// Returns sort order associated with this column order.
    pub fn sort_order(&self) -> SortOrder {
        match *self {
            ColumnOrder::TYPE_DEFINED_ORDER(order) => order,
            ColumnOrder::IEEE_754_TOTAL_ORDER => SortOrder::TOTAL_ORDER,
            ColumnOrder::INT96_TIMESTAMP_ORDER => SortOrder::INT96_TIMESTAMP,
            ColumnOrder::UNDEFINED => SortOrder::SIGNED,
            ColumnOrder::UNKNOWN => SortOrder::UNDEFINED,
        }
    }
}

impl<'a, R: ThriftCompactInputProtocol<'a>> ReadThrift<'a, R> for ColumnOrder {
    fn read_thrift(prot: &mut R) -> Result<Self> {
        let field_ident = prot.read_field_begin(0)?;
        if field_ident.field_type == FieldType::Stop {
            return Err(general_err!("Received empty union from remote ColumnOrder"));
        }
        let ret = match field_ident.id {
            1 => {
                // NOTE: the sort order needs to be set correctly after parsing.
                prot.skip_empty_struct()?;
                Self::TYPE_DEFINED_ORDER(SortOrder::SIGNED)
            }
            2 => {
                prot.skip_empty_struct()?;
                Self::IEEE_754_TOTAL_ORDER
            }
            3 => {
                prot.skip_empty_struct()?;
                Self::INT96_TIMESTAMP_ORDER
            }
            _ => {
                prot.skip(field_ident.field_type)?;
                Self::UNKNOWN
            }
        };
        let field_ident = prot.read_field_begin(field_ident.id)?;
        if field_ident.field_type != FieldType::Stop {
            return Err(general_err!(
                "Received multiple fields for union from remote ColumnOrder"
            ));
        }
        Ok(ret)
    }
}

impl WriteThrift for ColumnOrder {
    const ELEMENT_TYPE: ElementType = ElementType::Struct;

    fn write_thrift<W: Write>(&self, writer: &mut ThriftCompactOutputProtocol<W>) -> Result<()> {
        match *self {
            Self::TYPE_DEFINED_ORDER(_) => {
                writer.write_field_begin(FieldType::Struct, 1, 0)?;
                writer.write_struct_end()?;
            }
            Self::IEEE_754_TOTAL_ORDER => {
                writer.write_field_begin(FieldType::Struct, 2, 0)?;
                writer.write_struct_end()?;
            }
            Self::INT96_TIMESTAMP_ORDER => {
                writer.write_field_begin(FieldType::Struct, 3, 0)?;
                writer.write_struct_end()?;
            }
            _ => return Err(general_err!("Attempt to write undefined ColumnOrder")),
        }
        // write end of struct for this union
        writer.write_struct_end()
    }
}

// ----------------------------------------------------------------------
// Display handlers

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

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

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

// ----------------------------------------------------------------------
// LogicalType <=> ConvertedType conversion

// Note: To prevent type loss when converting from ConvertedType to LogicalType,
// the conversion from ConvertedType -> LogicalType is not implemented.
// Such type loss includes:
// - Not knowing the decimal scale and precision of ConvertedType
// - Time and timestamp nanosecond precision, that is not supported in ConvertedType.

impl From<Option<LogicalType>> for ConvertedType {
    fn from(value: Option<LogicalType>) -> Self {
        match value {
            Some(value) => match value {
                LogicalType::String => ConvertedType::UTF8,
                LogicalType::Map => ConvertedType::MAP,
                LogicalType::List => ConvertedType::LIST,
                LogicalType::Enum => ConvertedType::ENUM,
                LogicalType::Decimal { .. } => ConvertedType::DECIMAL,
                LogicalType::Date => ConvertedType::DATE,
                LogicalType::Time(time) => match time.unit {
                    TimeUnit::MILLIS => ConvertedType::TIME_MILLIS,
                    TimeUnit::MICROS => ConvertedType::TIME_MICROS,
                    TimeUnit::NANOS => ConvertedType::NONE,
                },
                LogicalType::Timestamp(time) => match time.unit {
                    TimeUnit::MILLIS => ConvertedType::TIMESTAMP_MILLIS,
                    TimeUnit::MICROS => ConvertedType::TIMESTAMP_MICROS,
                    TimeUnit::NANOS => ConvertedType::NONE,
                },
                LogicalType::Integer(int_type) => match (int_type.bit_width, int_type.is_signed) {
                    (8, true) => ConvertedType::INT_8,
                    (16, true) => ConvertedType::INT_16,
                    (32, true) => ConvertedType::INT_32,
                    (64, true) => ConvertedType::INT_64,
                    (8, false) => ConvertedType::UINT_8,
                    (16, false) => ConvertedType::UINT_16,
                    (32, false) => ConvertedType::UINT_32,
                    (64, false) => ConvertedType::UINT_64,
                    (bit_width, is_signed) => panic!(
                        "Integer type bit_width={bit_width}, signed={is_signed} is not supported"
                    ),
                },
                LogicalType::Json => ConvertedType::JSON,
                LogicalType::Bson => ConvertedType::BSON,
                LogicalType::Uuid
                | LogicalType::Float16
                | LogicalType::Variant(_)
                | LogicalType::Geometry(_)
                | LogicalType::Geography(_)
                | LogicalType::_Unknown { .. }
                | LogicalType::Unknown => ConvertedType::NONE,
            },
            None => ConvertedType::NONE,
        }
    }
}

// ----------------------------------------------------------------------
// String conversions for schema parsing.

impl str::FromStr for Repetition {
    type Err = ParquetError;

    fn from_str(s: &str) -> Result<Self> {
        match s {
            "REQUIRED" => Ok(Repetition::REQUIRED),
            "OPTIONAL" => Ok(Repetition::OPTIONAL),
            "REPEATED" => Ok(Repetition::REPEATED),
            other => Err(general_err!("Invalid parquet repetition {}", other)),
        }
    }
}

impl str::FromStr for Type {
    type Err = ParquetError;

    fn from_str(s: &str) -> Result<Self> {
        match s {
            "BOOLEAN" => Ok(Type::BOOLEAN),
            "INT32" => Ok(Type::INT32),
            "INT64" => Ok(Type::INT64),
            "INT96" => Ok(Type::INT96),
            "FLOAT" => Ok(Type::FLOAT),
            "DOUBLE" => Ok(Type::DOUBLE),
            "BYTE_ARRAY" | "BINARY" => Ok(Type::BYTE_ARRAY),
            "FIXED_LEN_BYTE_ARRAY" => Ok(Type::FIXED_LEN_BYTE_ARRAY),
            other => Err(general_err!("Invalid parquet type {}", other)),
        }
    }
}

impl str::FromStr for ConvertedType {
    type Err = ParquetError;

    fn from_str(s: &str) -> Result<Self> {
        match s {
            "NONE" => Ok(ConvertedType::NONE),
            "UTF8" => Ok(ConvertedType::UTF8),
            "MAP" => Ok(ConvertedType::MAP),
            "MAP_KEY_VALUE" => Ok(ConvertedType::MAP_KEY_VALUE),
            "LIST" => Ok(ConvertedType::LIST),
            "ENUM" => Ok(ConvertedType::ENUM),
            "DECIMAL" => Ok(ConvertedType::DECIMAL),
            "DATE" => Ok(ConvertedType::DATE),
            "TIME_MILLIS" => Ok(ConvertedType::TIME_MILLIS),
            "TIME_MICROS" => Ok(ConvertedType::TIME_MICROS),
            "TIMESTAMP_MILLIS" => Ok(ConvertedType::TIMESTAMP_MILLIS),
            "TIMESTAMP_MICROS" => Ok(ConvertedType::TIMESTAMP_MICROS),
            "UINT_8" => Ok(ConvertedType::UINT_8),
            "UINT_16" => Ok(ConvertedType::UINT_16),
            "UINT_32" => Ok(ConvertedType::UINT_32),
            "UINT_64" => Ok(ConvertedType::UINT_64),
            "INT_8" => Ok(ConvertedType::INT_8),
            "INT_16" => Ok(ConvertedType::INT_16),
            "INT_32" => Ok(ConvertedType::INT_32),
            "INT_64" => Ok(ConvertedType::INT_64),
            "JSON" => Ok(ConvertedType::JSON),
            "BSON" => Ok(ConvertedType::BSON),
            "INTERVAL" => Ok(ConvertedType::INTERVAL),
            other => Err(general_err!("Invalid parquet converted type {}", other)),
        }
    }
}

impl str::FromStr for LogicalType {
    type Err = ParquetError;

    fn from_str(s: &str) -> Result<Self> {
        match s {
            // The type is a placeholder that gets updated elsewhere
            "INTEGER" => Ok(LogicalType::integer(8, false)),
            "MAP" => Ok(LogicalType::Map),
            "LIST" => Ok(LogicalType::List),
            "ENUM" => Ok(LogicalType::Enum),
            "DECIMAL" => Ok(LogicalType::decimal(-1, -1)),
            "DATE" => Ok(LogicalType::Date),
            "TIME" => Ok(LogicalType::time(false, TimeUnit::MILLIS)),
            "TIMESTAMP" => Ok(LogicalType::timestamp(false, TimeUnit::MILLIS)),
            "STRING" => Ok(LogicalType::String),
            "JSON" => Ok(LogicalType::Json),
            "BSON" => Ok(LogicalType::Bson),
            "UUID" => Ok(LogicalType::Uuid),
            "UNKNOWN" => Ok(LogicalType::Unknown),
            "INTERVAL" => Err(general_err!(
                "Interval parquet logical type not yet supported"
            )),
            "FLOAT16" => Ok(LogicalType::Float16),
            "VARIANT" => Ok(LogicalType::variant(None)),
            "GEOMETRY" => Ok(LogicalType::geometry(None)),
            "GEOGRAPHY" => Ok(LogicalType::geography(
                None,
                Some(EdgeInterpolationAlgorithm::SPHERICAL),
            )),
            other => Err(general_err!("Invalid parquet logical type {}", other)),
        }
    }
}

#[cfg(test)]
#[allow(deprecated)] // allow BIT_PACKED encoding for the whole test module
mod tests {
    use super::*;
    use crate::parquet_thrift::{ThriftSliceInputProtocol, tests::test_roundtrip};

    #[test]
    fn test_display_type() {
        assert_eq!(Type::BOOLEAN.to_string(), "BOOLEAN");
        assert_eq!(Type::INT32.to_string(), "INT32");
        assert_eq!(Type::INT64.to_string(), "INT64");
        assert_eq!(Type::INT96.to_string(), "INT96");
        assert_eq!(Type::FLOAT.to_string(), "FLOAT");
        assert_eq!(Type::DOUBLE.to_string(), "DOUBLE");
        assert_eq!(Type::BYTE_ARRAY.to_string(), "BYTE_ARRAY");
        assert_eq!(
            Type::FIXED_LEN_BYTE_ARRAY.to_string(),
            "FIXED_LEN_BYTE_ARRAY"
        );
    }

    #[test]
    fn test_from_string_into_type() {
        assert_eq!(
            Type::BOOLEAN.to_string().parse::<Type>().unwrap(),
            Type::BOOLEAN
        );
        assert_eq!(
            Type::INT32.to_string().parse::<Type>().unwrap(),
            Type::INT32
        );
        assert_eq!(
            Type::INT64.to_string().parse::<Type>().unwrap(),
            Type::INT64
        );
        assert_eq!(
            Type::INT96.to_string().parse::<Type>().unwrap(),
            Type::INT96
        );
        assert_eq!(
            Type::FLOAT.to_string().parse::<Type>().unwrap(),
            Type::FLOAT
        );
        assert_eq!(
            Type::DOUBLE.to_string().parse::<Type>().unwrap(),
            Type::DOUBLE
        );
        assert_eq!(
            Type::BYTE_ARRAY.to_string().parse::<Type>().unwrap(),
            Type::BYTE_ARRAY
        );
        assert_eq!("BINARY".parse::<Type>().unwrap(), Type::BYTE_ARRAY);
        assert_eq!(
            Type::FIXED_LEN_BYTE_ARRAY
                .to_string()
                .parse::<Type>()
                .unwrap(),
            Type::FIXED_LEN_BYTE_ARRAY
        );
    }

    #[test]
    fn test_converted_type_roundtrip() {
        test_roundtrip(ConvertedType::UTF8);
        test_roundtrip(ConvertedType::MAP);
        test_roundtrip(ConvertedType::MAP_KEY_VALUE);
        test_roundtrip(ConvertedType::LIST);
        test_roundtrip(ConvertedType::ENUM);
        test_roundtrip(ConvertedType::DECIMAL);
        test_roundtrip(ConvertedType::DATE);
        test_roundtrip(ConvertedType::TIME_MILLIS);
        test_roundtrip(ConvertedType::TIME_MICROS);
        test_roundtrip(ConvertedType::TIMESTAMP_MILLIS);
        test_roundtrip(ConvertedType::TIMESTAMP_MICROS);
        test_roundtrip(ConvertedType::UINT_8);
        test_roundtrip(ConvertedType::UINT_16);
        test_roundtrip(ConvertedType::UINT_32);
        test_roundtrip(ConvertedType::UINT_64);
        test_roundtrip(ConvertedType::INT_8);
        test_roundtrip(ConvertedType::INT_16);
        test_roundtrip(ConvertedType::INT_32);
        test_roundtrip(ConvertedType::INT_64);
        test_roundtrip(ConvertedType::JSON);
        test_roundtrip(ConvertedType::BSON);
        test_roundtrip(ConvertedType::INTERVAL);
    }

    #[test]
    fn test_read_invalid_converted_type() {
        let mut prot = ThriftSliceInputProtocol::new(&[0x7eu8]);
        let res = ConvertedType::read_thrift(&mut prot);
        assert!(res.is_err());
        assert_eq!(
            res.unwrap_err().to_string(),
            "Parquet error: Unexpected ConvertedType 63"
        );
    }

    #[test]
    fn test_display_converted_type() {
        assert_eq!(ConvertedType::NONE.to_string(), "NONE");
        assert_eq!(ConvertedType::UTF8.to_string(), "UTF8");
        assert_eq!(ConvertedType::MAP.to_string(), "MAP");
        assert_eq!(ConvertedType::MAP_KEY_VALUE.to_string(), "MAP_KEY_VALUE");
        assert_eq!(ConvertedType::LIST.to_string(), "LIST");
        assert_eq!(ConvertedType::ENUM.to_string(), "ENUM");
        assert_eq!(ConvertedType::DECIMAL.to_string(), "DECIMAL");
        assert_eq!(ConvertedType::DATE.to_string(), "DATE");
        assert_eq!(ConvertedType::TIME_MILLIS.to_string(), "TIME_MILLIS");
        assert_eq!(ConvertedType::DATE.to_string(), "DATE");
        assert_eq!(ConvertedType::TIME_MICROS.to_string(), "TIME_MICROS");
        assert_eq!(
            ConvertedType::TIMESTAMP_MILLIS.to_string(),
            "TIMESTAMP_MILLIS"
        );
        assert_eq!(
            ConvertedType::TIMESTAMP_MICROS.to_string(),
            "TIMESTAMP_MICROS"
        );
        assert_eq!(ConvertedType::UINT_8.to_string(), "UINT_8");
        assert_eq!(ConvertedType::UINT_16.to_string(), "UINT_16");
        assert_eq!(ConvertedType::UINT_32.to_string(), "UINT_32");
        assert_eq!(ConvertedType::UINT_64.to_string(), "UINT_64");
        assert_eq!(ConvertedType::INT_8.to_string(), "INT_8");
        assert_eq!(ConvertedType::INT_16.to_string(), "INT_16");
        assert_eq!(ConvertedType::INT_32.to_string(), "INT_32");
        assert_eq!(ConvertedType::INT_64.to_string(), "INT_64");
        assert_eq!(ConvertedType::JSON.to_string(), "JSON");
        assert_eq!(ConvertedType::BSON.to_string(), "BSON");
        assert_eq!(ConvertedType::INTERVAL.to_string(), "INTERVAL");
        assert_eq!(ConvertedType::DECIMAL.to_string(), "DECIMAL")
    }

    #[test]
    fn test_from_string_into_converted_type() {
        assert_eq!(
            ConvertedType::NONE
                .to_string()
                .parse::<ConvertedType>()
                .unwrap(),
            ConvertedType::NONE
        );
        assert_eq!(
            ConvertedType::UTF8
                .to_string()
                .parse::<ConvertedType>()
                .unwrap(),
            ConvertedType::UTF8
        );
        assert_eq!(
            ConvertedType::MAP
                .to_string()
                .parse::<ConvertedType>()
                .unwrap(),
            ConvertedType::MAP
        );
        assert_eq!(
            ConvertedType::MAP_KEY_VALUE
                .to_string()
                .parse::<ConvertedType>()
                .unwrap(),
            ConvertedType::MAP_KEY_VALUE
        );
        assert_eq!(
            ConvertedType::LIST
                .to_string()
                .parse::<ConvertedType>()
                .unwrap(),
            ConvertedType::LIST
        );
        assert_eq!(
            ConvertedType::ENUM
                .to_string()
                .parse::<ConvertedType>()
                .unwrap(),
            ConvertedType::ENUM
        );
        assert_eq!(
            ConvertedType::DECIMAL
                .to_string()
                .parse::<ConvertedType>()
                .unwrap(),
            ConvertedType::DECIMAL
        );
        assert_eq!(
            ConvertedType::DATE
                .to_string()
                .parse::<ConvertedType>()
                .unwrap(),
            ConvertedType::DATE
        );
        assert_eq!(
            ConvertedType::TIME_MILLIS
                .to_string()
                .parse::<ConvertedType>()
                .unwrap(),
            ConvertedType::TIME_MILLIS
        );
        assert_eq!(
            ConvertedType::TIME_MICROS
                .to_string()
                .parse::<ConvertedType>()
                .unwrap(),
            ConvertedType::TIME_MICROS
        );
        assert_eq!(
            ConvertedType::TIMESTAMP_MILLIS
                .to_string()
                .parse::<ConvertedType>()
                .unwrap(),
            ConvertedType::TIMESTAMP_MILLIS
        );
        assert_eq!(
            ConvertedType::TIMESTAMP_MICROS
                .to_string()
                .parse::<ConvertedType>()
                .unwrap(),
            ConvertedType::TIMESTAMP_MICROS
        );
        assert_eq!(
            ConvertedType::UINT_8
                .to_string()
                .parse::<ConvertedType>()
                .unwrap(),
            ConvertedType::UINT_8
        );
        assert_eq!(
            ConvertedType::UINT_16
                .to_string()
                .parse::<ConvertedType>()
                .unwrap(),
            ConvertedType::UINT_16
        );
        assert_eq!(
            ConvertedType::UINT_32
                .to_string()
                .parse::<ConvertedType>()
                .unwrap(),
            ConvertedType::UINT_32
        );
        assert_eq!(
            ConvertedType::UINT_64
                .to_string()
                .parse::<ConvertedType>()
                .unwrap(),
            ConvertedType::UINT_64
        );
        assert_eq!(
            ConvertedType::INT_8
                .to_string()
                .parse::<ConvertedType>()
                .unwrap(),
            ConvertedType::INT_8
        );
        assert_eq!(
            ConvertedType::INT_16
                .to_string()
                .parse::<ConvertedType>()
                .unwrap(),
            ConvertedType::INT_16
        );
        assert_eq!(
            ConvertedType::INT_32
                .to_string()
                .parse::<ConvertedType>()
                .unwrap(),
            ConvertedType::INT_32
        );
        assert_eq!(
            ConvertedType::INT_64
                .to_string()
                .parse::<ConvertedType>()
                .unwrap(),
            ConvertedType::INT_64
        );
        assert_eq!(
            ConvertedType::JSON
                .to_string()
                .parse::<ConvertedType>()
                .unwrap(),
            ConvertedType::JSON
        );
        assert_eq!(
            ConvertedType::BSON
                .to_string()
                .parse::<ConvertedType>()
                .unwrap(),
            ConvertedType::BSON
        );
        assert_eq!(
            ConvertedType::INTERVAL
                .to_string()
                .parse::<ConvertedType>()
                .unwrap(),
            ConvertedType::INTERVAL
        );
        assert_eq!(
            ConvertedType::DECIMAL
                .to_string()
                .parse::<ConvertedType>()
                .unwrap(),
            ConvertedType::DECIMAL
        )
    }

    #[test]
    fn test_logical_to_converted_type() {
        let logical_none: Option<LogicalType> = None;
        assert_eq!(ConvertedType::from(logical_none), ConvertedType::NONE);
        assert_eq!(
            ConvertedType::from(Some(LogicalType::decimal(5, 20))),
            ConvertedType::DECIMAL
        );
        assert_eq!(
            ConvertedType::from(Some(LogicalType::Bson)),
            ConvertedType::BSON
        );
        assert_eq!(
            ConvertedType::from(Some(LogicalType::Json)),
            ConvertedType::JSON
        );
        assert_eq!(
            ConvertedType::from(Some(LogicalType::String)),
            ConvertedType::UTF8
        );
        assert_eq!(
            ConvertedType::from(Some(LogicalType::Date)),
            ConvertedType::DATE
        );
        assert_eq!(
            ConvertedType::from(Some(LogicalType::time(true, TimeUnit::MILLIS))),
            ConvertedType::TIME_MILLIS
        );
        assert_eq!(
            ConvertedType::from(Some(LogicalType::time(true, TimeUnit::MICROS))),
            ConvertedType::TIME_MICROS
        );
        assert_eq!(
            ConvertedType::from(Some(LogicalType::time(false, TimeUnit::NANOS))),
            ConvertedType::NONE
        );
        assert_eq!(
            ConvertedType::from(Some(LogicalType::timestamp(true, TimeUnit::MILLIS))),
            ConvertedType::TIMESTAMP_MILLIS
        );
        assert_eq!(
            ConvertedType::from(Some(LogicalType::timestamp(false, TimeUnit::MICROS))),
            ConvertedType::TIMESTAMP_MICROS
        );
        assert_eq!(
            ConvertedType::from(Some(LogicalType::timestamp(false, TimeUnit::NANOS))),
            ConvertedType::NONE
        );
        assert_eq!(
            ConvertedType::from(Some(LogicalType::integer(8, false))),
            ConvertedType::UINT_8
        );
        assert_eq!(
            ConvertedType::from(Some(LogicalType::integer(8, true))),
            ConvertedType::INT_8
        );
        assert_eq!(
            ConvertedType::from(Some(LogicalType::integer(16, false))),
            ConvertedType::UINT_16
        );
        assert_eq!(
            ConvertedType::from(Some(LogicalType::integer(16, true))),
            ConvertedType::INT_16
        );
        assert_eq!(
            ConvertedType::from(Some(LogicalType::integer(32, false))),
            ConvertedType::UINT_32
        );
        assert_eq!(
            ConvertedType::from(Some(LogicalType::integer(32, true))),
            ConvertedType::INT_32
        );
        assert_eq!(
            ConvertedType::from(Some(LogicalType::integer(64, false))),
            ConvertedType::UINT_64
        );
        assert_eq!(
            ConvertedType::from(Some(LogicalType::integer(64, true))),
            ConvertedType::INT_64
        );
        assert_eq!(
            ConvertedType::from(Some(LogicalType::List)),
            ConvertedType::LIST
        );
        assert_eq!(
            ConvertedType::from(Some(LogicalType::Map)),
            ConvertedType::MAP
        );
        assert_eq!(
            ConvertedType::from(Some(LogicalType::Uuid)),
            ConvertedType::NONE
        );
        assert_eq!(
            ConvertedType::from(Some(LogicalType::Enum)),
            ConvertedType::ENUM
        );
        assert_eq!(
            ConvertedType::from(Some(LogicalType::Float16)),
            ConvertedType::NONE
        );
        assert_eq!(
            ConvertedType::from(Some(LogicalType::variant(None))),
            ConvertedType::NONE
        );
        assert_eq!(
            ConvertedType::from(Some(LogicalType::geometry(None))),
            ConvertedType::NONE
        );
        assert_eq!(
            ConvertedType::from(Some(LogicalType::geography(None, Some(Default::default())))),
            ConvertedType::NONE
        );
        assert_eq!(
            ConvertedType::from(Some(LogicalType::Unknown)),
            ConvertedType::NONE
        );
    }

    #[test]
    fn test_logical_type_roundtrip() {
        test_roundtrip(LogicalType::String);
        test_roundtrip(LogicalType::Map);
        test_roundtrip(LogicalType::List);
        test_roundtrip(LogicalType::Enum);
        test_roundtrip(LogicalType::decimal(0, 20));
        test_roundtrip(LogicalType::Date);
        test_roundtrip(LogicalType::time(true, TimeUnit::MICROS));
        test_roundtrip(LogicalType::time(false, TimeUnit::MILLIS));
        test_roundtrip(LogicalType::time(false, TimeUnit::NANOS));
        test_roundtrip(LogicalType::timestamp(false, TimeUnit::MICROS));
        test_roundtrip(LogicalType::timestamp(true, TimeUnit::MILLIS));
        test_roundtrip(LogicalType::timestamp(true, TimeUnit::NANOS));
        test_roundtrip(LogicalType::integer(8, true));
        test_roundtrip(LogicalType::integer(16, false));
        test_roundtrip(LogicalType::integer(32, true));
        test_roundtrip(LogicalType::integer(64, false));
        test_roundtrip(LogicalType::Json);
        test_roundtrip(LogicalType::Bson);
        test_roundtrip(LogicalType::Uuid);
        test_roundtrip(LogicalType::Float16);
        test_roundtrip(LogicalType::variant(Some(1)));
        test_roundtrip(LogicalType::variant(None));
        test_roundtrip(LogicalType::geometry(Some("foo".to_owned())));
        test_roundtrip(LogicalType::geometry(None));
        test_roundtrip(LogicalType::geography(
            Some("foo".to_owned()),
            Some(EdgeInterpolationAlgorithm::ANDOYER),
        ));
        test_roundtrip(LogicalType::geography(
            None,
            Some(EdgeInterpolationAlgorithm::KARNEY),
        ));
        test_roundtrip(LogicalType::geography(
            Some("foo".to_owned()),
            Some(EdgeInterpolationAlgorithm::SPHERICAL),
        ));
        test_roundtrip(LogicalType::geography(
            None,
            Some(EdgeInterpolationAlgorithm::SPHERICAL),
        ));
    }

    #[test]
    fn test_display_repetition() {
        assert_eq!(Repetition::REQUIRED.to_string(), "REQUIRED");
        assert_eq!(Repetition::OPTIONAL.to_string(), "OPTIONAL");
        assert_eq!(Repetition::REPEATED.to_string(), "REPEATED");
    }

    #[test]
    fn test_from_string_into_repetition() {
        assert_eq!(
            Repetition::REQUIRED
                .to_string()
                .parse::<Repetition>()
                .unwrap(),
            Repetition::REQUIRED
        );
        assert_eq!(
            Repetition::OPTIONAL
                .to_string()
                .parse::<Repetition>()
                .unwrap(),
            Repetition::OPTIONAL
        );
        assert_eq!(
            Repetition::REPEATED
                .to_string()
                .parse::<Repetition>()
                .unwrap(),
            Repetition::REPEATED
        );
    }

    #[test]
    fn test_display_encoding() {
        assert_eq!(Encoding::PLAIN.to_string(), "PLAIN");
        assert_eq!(Encoding::PLAIN_DICTIONARY.to_string(), "PLAIN_DICTIONARY");
        assert_eq!(Encoding::RLE.to_string(), "RLE");
        assert_eq!(Encoding::BIT_PACKED.to_string(), "BIT_PACKED");
        assert_eq!(
            Encoding::DELTA_BINARY_PACKED.to_string(),
            "DELTA_BINARY_PACKED"
        );
        assert_eq!(
            Encoding::DELTA_LENGTH_BYTE_ARRAY.to_string(),
            "DELTA_LENGTH_BYTE_ARRAY"
        );
        assert_eq!(Encoding::DELTA_BYTE_ARRAY.to_string(), "DELTA_BYTE_ARRAY");
        assert_eq!(Encoding::RLE_DICTIONARY.to_string(), "RLE_DICTIONARY");
    }

    #[test]
    fn test_compression_conversion() {
        assert_eq!(
            CompressionCodec::from(Compression::UNCOMPRESSED),
            CompressionCodec::UNCOMPRESSED
        );
        assert_eq!(
            CompressionCodec::from(Compression::SNAPPY),
            CompressionCodec::SNAPPY
        );
        assert_eq!(
            CompressionCodec::from(Compression::GZIP(Default::default())),
            CompressionCodec::GZIP
        );
        assert_eq!(
            CompressionCodec::from(Compression::LZO),
            CompressionCodec::LZO
        );
        assert_eq!(
            CompressionCodec::from(Compression::BROTLI(Default::default())),
            CompressionCodec::BROTLI
        );
        assert_eq!(
            CompressionCodec::from(Compression::LZ4),
            CompressionCodec::LZ4
        );
        assert_eq!(
            CompressionCodec::from(Compression::ZSTD(Default::default())),
            CompressionCodec::ZSTD
        );
        assert_eq!(
            CompressionCodec::from(Compression::LZ4_RAW),
            CompressionCodec::LZ4_RAW
        );

        assert_eq!(
            Compression::from(CompressionCodec::UNCOMPRESSED),
            Compression::UNCOMPRESSED
        );
        assert_eq!(
            Compression::from(CompressionCodec::SNAPPY),
            Compression::SNAPPY
        );
        assert_eq!(
            Compression::from(CompressionCodec::GZIP),
            Compression::GZIP(Default::default())
        );
        assert_eq!(Compression::from(CompressionCodec::LZO), Compression::LZO);
        assert_eq!(
            Compression::from(CompressionCodec::BROTLI),
            Compression::BROTLI(Default::default())
        );
        assert_eq!(Compression::from(CompressionCodec::LZ4), Compression::LZ4);
        assert_eq!(
            Compression::from(CompressionCodec::ZSTD),
            Compression::ZSTD(Default::default())
        );
        assert_eq!(
            Compression::from(CompressionCodec::LZ4_RAW),
            Compression::LZ4_RAW
        );
    }

    #[test]
    fn test_display_compression() {
        assert_eq!(Compression::UNCOMPRESSED.to_string(), "UNCOMPRESSED");
        assert_eq!(Compression::SNAPPY.to_string(), "SNAPPY");
        assert_eq!(
            Compression::GZIP(Default::default()).to_string(),
            "GZIP(GzipLevel(6))"
        );
        assert_eq!(Compression::LZO.to_string(), "LZO");
        assert_eq!(
            Compression::BROTLI(Default::default()).to_string(),
            "BROTLI(BrotliLevel(1))"
        );
        assert_eq!(Compression::LZ4.to_string(), "LZ4");
        assert_eq!(
            Compression::ZSTD(Default::default()).to_string(),
            "ZSTD(ZstdLevel(1))"
        );
    }

    #[test]
    fn test_display_page_type() {
        assert_eq!(PageType::DATA_PAGE.to_string(), "DATA_PAGE");
        assert_eq!(PageType::INDEX_PAGE.to_string(), "INDEX_PAGE");
        assert_eq!(PageType::DICTIONARY_PAGE.to_string(), "DICTIONARY_PAGE");
        assert_eq!(PageType::DATA_PAGE_V2.to_string(), "DATA_PAGE_V2");
    }

    #[test]
    fn test_display_sort_order() {
        assert_eq!(SortOrder::SIGNED.to_string(), "SIGNED");
        assert_eq!(SortOrder::UNSIGNED.to_string(), "UNSIGNED");
        assert_eq!(SortOrder::UNDEFINED.to_string(), "UNDEFINED");
        assert_eq!(SortOrder::TOTAL_ORDER.to_string(), "TOTAL_ORDER");
        assert_eq!(SortOrder::INT96_TIMESTAMP.to_string(), "INT96_TIMESTAMP");
    }

    #[test]
    fn test_display_column_order() {
        assert_eq!(
            ColumnOrder::TYPE_DEFINED_ORDER(SortOrder::SIGNED).to_string(),
            "TYPE_DEFINED_ORDER(SIGNED)"
        );
        assert_eq!(
            ColumnOrder::TYPE_DEFINED_ORDER(SortOrder::UNSIGNED).to_string(),
            "TYPE_DEFINED_ORDER(UNSIGNED)"
        );
        assert_eq!(
            ColumnOrder::TYPE_DEFINED_ORDER(SortOrder::UNDEFINED).to_string(),
            "TYPE_DEFINED_ORDER(UNDEFINED)"
        );
        assert_eq!(
            ColumnOrder::IEEE_754_TOTAL_ORDER.to_string(),
            "IEEE_754_TOTAL_ORDER"
        );
        assert_eq!(
            ColumnOrder::INT96_TIMESTAMP_ORDER.to_string(),
            "INT96_TIMESTAMP_ORDER"
        );
        assert_eq!(ColumnOrder::UNDEFINED.to_string(), "UNDEFINED");
    }

    #[test]
    fn test_column_order_roundtrip() {
        // SortOrder::SIGNED is the default on read.
        test_roundtrip(ColumnOrder::TYPE_DEFINED_ORDER(SortOrder::SIGNED))
    }

    #[test]
    fn test_column_order_get_logical_type_sort_order() {
        // Helper to check the order in a list of values.
        // Only logical type is checked.
        fn check_sort_order(types: Vec<LogicalType>, expected_order: SortOrder) {
            for tpe in types {
                assert_eq!(
                    ColumnOrder::column_order_for_type(
                        Some(&tpe),
                        ConvertedType::NONE,
                        Type::BYTE_ARRAY
                    )
                    .sort_order(),
                    expected_order
                );
            }
        }

        // Unsigned comparison (physical type does not matter)
        let unsigned = vec![
            LogicalType::String,
            LogicalType::Json,
            LogicalType::Bson,
            LogicalType::Enum,
            LogicalType::Uuid,
            LogicalType::integer(8, false),
            LogicalType::integer(16, false),
            LogicalType::integer(32, false),
            LogicalType::integer(64, false),
        ];
        check_sort_order(unsigned, SortOrder::UNSIGNED);

        // Signed comparison (physical type does not matter)
        let signed = vec![
            LogicalType::integer(8, true),
            LogicalType::integer(16, true),
            LogicalType::integer(32, true),
            LogicalType::integer(64, true),
            LogicalType::decimal(20, 4),
            LogicalType::Date,
            LogicalType::time(false, TimeUnit::MILLIS),
            LogicalType::time(false, TimeUnit::MICROS),
            LogicalType::time(true, TimeUnit::NANOS),
            LogicalType::timestamp(false, TimeUnit::MILLIS),
            LogicalType::timestamp(false, TimeUnit::MICROS),
            LogicalType::timestamp(true, TimeUnit::NANOS),
        ];
        check_sort_order(signed, SortOrder::SIGNED);

        let float = vec![LogicalType::Float16];
        check_sort_order(float, SortOrder::TOTAL_ORDER);

        // Undefined comparison
        let undefined = vec![
            LogicalType::List,
            LogicalType::Map,
            LogicalType::variant(None),
            LogicalType::geometry(None),
            LogicalType::geography(None, Some(Default::default())),
        ];
        check_sort_order(undefined, SortOrder::UNDEFINED);
    }

    #[test]
    fn test_column_order_get_converted_type_sort_order() {
        // Helper to check the order in a list of values.
        // Only converted type is checked.
        fn check_sort_order(types: Vec<ConvertedType>, expected_order: SortOrder) {
            for tpe in types {
                assert_eq!(
                    ColumnOrder::column_order_for_type(None, tpe, Type::BYTE_ARRAY).sort_order(),
                    expected_order
                );
            }
        }

        // Unsigned comparison (physical type does not matter)
        let unsigned = vec![
            ConvertedType::UTF8,
            ConvertedType::JSON,
            ConvertedType::BSON,
            ConvertedType::ENUM,
            ConvertedType::UINT_8,
            ConvertedType::UINT_16,
            ConvertedType::UINT_32,
            ConvertedType::UINT_64,
        ];
        check_sort_order(unsigned, SortOrder::UNSIGNED);

        // Signed comparison (physical type does not matter)
        let signed = vec![
            ConvertedType::INT_8,
            ConvertedType::INT_16,
            ConvertedType::INT_32,
            ConvertedType::INT_64,
            ConvertedType::DECIMAL,
            ConvertedType::DATE,
            ConvertedType::TIME_MILLIS,
            ConvertedType::TIME_MICROS,
            ConvertedType::TIMESTAMP_MILLIS,
            ConvertedType::TIMESTAMP_MICROS,
        ];
        check_sort_order(signed, SortOrder::SIGNED);

        // Undefined comparison
        let undefined = vec![
            ConvertedType::LIST,
            ConvertedType::MAP,
            ConvertedType::MAP_KEY_VALUE,
            ConvertedType::INTERVAL,
        ];
        check_sort_order(undefined, SortOrder::UNDEFINED);

        // Check None logical type
        // This should return a sort order for byte array type.
        check_sort_order(vec![ConvertedType::NONE], SortOrder::UNSIGNED);
    }

    #[test]
    fn test_column_order_get_default_sort_order() {
        // Comparison based on physical type
        assert_eq!(
            ColumnOrder::get_default_sort_order(Type::BOOLEAN, true),
            SortOrder::UNSIGNED
        );
        assert_eq!(
            ColumnOrder::get_default_sort_order(Type::INT32, true),
            SortOrder::SIGNED
        );
        assert_eq!(
            ColumnOrder::get_default_sort_order(Type::INT64, true),
            SortOrder::SIGNED
        );
        assert_eq!(
            ColumnOrder::get_default_sort_order(Type::INT96, true),
            SortOrder::UNDEFINED
        );
        assert_eq!(
            ColumnOrder::get_default_sort_order(Type::INT96, false),
            SortOrder::INT96_TIMESTAMP
        );
        assert_eq!(
            ColumnOrder::get_default_sort_order(Type::FLOAT, false),
            SortOrder::TOTAL_ORDER
        );
        assert_eq!(
            ColumnOrder::get_default_sort_order(Type::DOUBLE, false),
            SortOrder::TOTAL_ORDER
        );
        assert_eq!(
            ColumnOrder::get_default_sort_order(Type::FLOAT, true),
            SortOrder::SIGNED
        );
        assert_eq!(
            ColumnOrder::get_default_sort_order(Type::DOUBLE, true),
            SortOrder::SIGNED
        );
        assert_eq!(
            ColumnOrder::get_default_sort_order(Type::BYTE_ARRAY, true),
            SortOrder::UNSIGNED
        );
        assert_eq!(
            ColumnOrder::get_default_sort_order(Type::FIXED_LEN_BYTE_ARRAY, true),
            SortOrder::UNSIGNED
        );
    }

    #[test]
    fn test_column_order_sort_order() {
        assert_eq!(
            ColumnOrder::TYPE_DEFINED_ORDER(SortOrder::SIGNED).sort_order(),
            SortOrder::SIGNED
        );
        assert_eq!(
            ColumnOrder::TYPE_DEFINED_ORDER(SortOrder::UNSIGNED).sort_order(),
            SortOrder::UNSIGNED
        );
        assert_eq!(
            ColumnOrder::TYPE_DEFINED_ORDER(SortOrder::UNDEFINED).sort_order(),
            SortOrder::UNDEFINED
        );
        assert_eq!(
            ColumnOrder::IEEE_754_TOTAL_ORDER.sort_order(),
            SortOrder::TOTAL_ORDER
        );
        assert_eq!(
            ColumnOrder::INT96_TIMESTAMP_ORDER.sort_order(),
            SortOrder::INT96_TIMESTAMP
        );
        assert_eq!(ColumnOrder::UNDEFINED.sort_order(), SortOrder::SIGNED);
    }

    #[test]
    fn test_parse_encoding() {
        let mut encoding: Encoding = "PLAIN".parse().unwrap();
        assert_eq!(encoding, Encoding::PLAIN);
        encoding = "PLAIN_DICTIONARY".parse().unwrap();
        assert_eq!(encoding, Encoding::PLAIN_DICTIONARY);
        encoding = "RLE".parse().unwrap();
        assert_eq!(encoding, Encoding::RLE);
        encoding = "BIT_PACKED".parse().unwrap();
        assert_eq!(encoding, Encoding::BIT_PACKED);
        encoding = "DELTA_BINARY_PACKED".parse().unwrap();
        assert_eq!(encoding, Encoding::DELTA_BINARY_PACKED);
        encoding = "DELTA_LENGTH_BYTE_ARRAY".parse().unwrap();
        assert_eq!(encoding, Encoding::DELTA_LENGTH_BYTE_ARRAY);
        encoding = "DELTA_BYTE_ARRAY".parse().unwrap();
        assert_eq!(encoding, Encoding::DELTA_BYTE_ARRAY);
        encoding = "RLE_DICTIONARY".parse().unwrap();
        assert_eq!(encoding, Encoding::RLE_DICTIONARY);
        encoding = "BYTE_STREAM_SPLIT".parse().unwrap();
        assert_eq!(encoding, Encoding::BYTE_STREAM_SPLIT);

        // test lowercase
        encoding = "byte_stream_split".parse().unwrap();
        assert_eq!(encoding, Encoding::BYTE_STREAM_SPLIT);

        // test unknown string
        match "plain_xxx".parse::<Encoding>() {
            Ok(e) => {
                panic!("Should not be able to parse {e:?}");
            }
            Err(e) => {
                assert_eq!(e.to_string(), "Parquet error: unknown encoding: plain_xxx");
            }
        }
    }

    #[test]
    fn test_parse_compression() {
        let mut compress: Compression = "snappy".parse().unwrap();
        assert_eq!(compress, Compression::SNAPPY);
        compress = "lzo".parse().unwrap();
        assert_eq!(compress, Compression::LZO);
        compress = "zstd(3)".parse().unwrap();
        assert_eq!(compress, Compression::ZSTD(ZstdLevel::try_new(3).unwrap()));
        compress = "zstd(-3)".parse().unwrap();
        assert_eq!(compress, Compression::ZSTD(ZstdLevel::try_new(-3).unwrap()));
        compress = "LZ4_RAW".parse().unwrap();
        assert_eq!(compress, Compression::LZ4_RAW);
        compress = "uncompressed".parse().unwrap();
        assert_eq!(compress, Compression::UNCOMPRESSED);
        compress = "snappy".parse().unwrap();
        assert_eq!(compress, Compression::SNAPPY);
        compress = "gzip(9)".parse().unwrap();
        assert_eq!(compress, Compression::GZIP(GzipLevel::try_new(9).unwrap()));
        compress = "lzo".parse().unwrap();
        assert_eq!(compress, Compression::LZO);
        compress = "brotli(3)".parse().unwrap();
        assert_eq!(
            compress,
            Compression::BROTLI(BrotliLevel::try_new(3).unwrap())
        );
        compress = "lz4".parse().unwrap();
        assert_eq!(compress, Compression::LZ4);

        // test unknown compression
        let mut err = "plain_xxx".parse::<Encoding>().unwrap_err();
        assert_eq!(
            err.to_string(),
            "Parquet error: unknown encoding: plain_xxx"
        );

        // test invalid compress level
        err = "gzip(-10)".parse::<Encoding>().unwrap_err();
        assert_eq!(
            err.to_string(),
            "Parquet error: unknown encoding: gzip(-10)"
        );
    }

    #[test]
    fn test_display_boundary_order() {
        assert_eq!(BoundaryOrder::ASCENDING.to_string(), "ASCENDING");
        assert_eq!(BoundaryOrder::DESCENDING.to_string(), "DESCENDING");
        assert_eq!(BoundaryOrder::UNORDERED.to_string(), "UNORDERED");
    }

    #[test]
    fn test_display_edge_algo() {
        assert_eq!(
            EdgeInterpolationAlgorithm::SPHERICAL.to_string(),
            "SPHERICAL"
        );
        assert_eq!(EdgeInterpolationAlgorithm::VINCENTY.to_string(), "VINCENTY");
        assert_eq!(EdgeInterpolationAlgorithm::THOMAS.to_string(), "THOMAS");
        assert_eq!(EdgeInterpolationAlgorithm::ANDOYER.to_string(), "ANDOYER");
        assert_eq!(EdgeInterpolationAlgorithm::KARNEY.to_string(), "KARNEY");
    }

    #[test]
    fn test_from_str_edge_algo() {
        assert_eq!(
            "spHErical".parse::<EdgeInterpolationAlgorithm>().unwrap(),
            EdgeInterpolationAlgorithm::SPHERICAL
        );
        assert_eq!(
            "vinceNTY".parse::<EdgeInterpolationAlgorithm>().unwrap(),
            EdgeInterpolationAlgorithm::VINCENTY
        );
        assert_eq!(
            "tHOmas".parse::<EdgeInterpolationAlgorithm>().unwrap(),
            EdgeInterpolationAlgorithm::THOMAS
        );
        assert_eq!(
            "anDOYEr".parse::<EdgeInterpolationAlgorithm>().unwrap(),
            EdgeInterpolationAlgorithm::ANDOYER
        );
        assert_eq!(
            "kaRNey".parse::<EdgeInterpolationAlgorithm>().unwrap(),
            EdgeInterpolationAlgorithm::KARNEY
        );
        assert!(
            "does not exist"
                .parse::<EdgeInterpolationAlgorithm>()
                .is_err()
        );
    }

    fn encodings_roundtrip(mut encodings: Vec<Encoding>) {
        encodings.sort();
        let mask = EncodingMask::new_from_encodings(encodings.iter());
        assert!(mask.all_set(encodings.iter()));
        let v = mask.encodings().collect::<Vec<_>>();
        assert_eq!(v, encodings);
    }

    #[test]
    fn test_encoding_roundtrip() {
        encodings_roundtrip(
            [
                Encoding::RLE,
                Encoding::PLAIN,
                Encoding::DELTA_BINARY_PACKED,
            ]
            .into(),
        );
        encodings_roundtrip([Encoding::RLE_DICTIONARY, Encoding::PLAIN_DICTIONARY].into());
        encodings_roundtrip([].into());
        let encodings = [
            Encoding::PLAIN,
            Encoding::BIT_PACKED,
            Encoding::RLE,
            Encoding::DELTA_BINARY_PACKED,
            Encoding::DELTA_BYTE_ARRAY,
            Encoding::DELTA_LENGTH_BYTE_ARRAY,
            Encoding::PLAIN_DICTIONARY,
            Encoding::RLE_DICTIONARY,
            Encoding::BYTE_STREAM_SPLIT,
        ];
        encodings_roundtrip(encodings.into());
    }

    #[test]
    fn test_invalid_encoding_mask() {
        // any set bits higher than the max should trigger an error
        let res = EncodingMask::try_new(-1);
        assert!(res.is_err());
        let err = res.unwrap_err();
        assert_eq!(
            err.to_string(),
            "Parquet error: Attempt to create invalid mask: 0xffffffff"
        );

        // test that GROUP_VAR_INT is disallowed
        let res = EncodingMask::try_new(2);
        assert!(res.is_err());
        let err = res.unwrap_err();
        assert_eq!(
            err.to_string(),
            "Parquet error: Attempt to create invalid mask: 0x2"
        );
    }

    #[test]
    fn test_encoding_mask_is_only() {
        let mask = EncodingMask::new_from_encodings([Encoding::PLAIN].iter());
        assert!(mask.is_only(Encoding::PLAIN));

        let mask =
            EncodingMask::new_from_encodings([Encoding::PLAIN, Encoding::PLAIN_DICTIONARY].iter());
        assert!(!mask.is_only(Encoding::PLAIN));
    }
}