You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@arrow.apache.org by et...@apache.org on 2025/08/11 22:23:59 UTC
(arrow-rs) branch gh5854_thrift_remodel updated: [thrift-remodel] Redo thrift enums and unions (#8072)
This is an automated email from the ASF dual-hosted git repository.
etseidl pushed a commit to branch gh5854_thrift_remodel
in repository https://gitbox.apache.org/repos/asf/arrow-rs.git
The following commit(s) were added to refs/heads/gh5854_thrift_remodel by this push:
new 50f33231b2 [thrift-remodel] Redo thrift enums and unions (#8072)
50f33231b2 is described below
commit 50f33231b200d9456de5e73e8a3c8aea51312854
Author: Ed Seidl <et...@users.noreply.github.com>
AuthorDate: Mon Aug 11 15:23:54 2025 -0700
[thrift-remodel] Redo thrift enums and unions (#8072)
# Which issue does this PR close?
**Note: this targets a feature branch, not main**
- Part of #5854.
- Related to #6129
# Rationale for this change
Next step.
# What changes are included in this PR?
Attempt to make use of macros originally developed by @jhorstmann to
implement some thrift enums and unions in `basic.rs`. Also adds yet
another spin on a thrift decoder based heavily on
`TCompactSliceInputProtocol`. The current approach is to use this new
decoder with `TryFrom` impls on the data structures in question.
This PR does not yet complete the process far enough to parse a parquet
footer, but I'm putting it out here to get early feedback on the design.
Some structures already deviate enough from their thrift counterparts
that the macro based parsing is not practical. Likewise, the macro
approach doesn't work for structs that need lifetime annotations
(necessary for reading binary fields as slices).
# Are these changes tested?
Not yet.
# Are there any user-facing changes?
Yes.
---
parquet/src/basic.rs | 940 +++++++++++++++++++--------------
parquet/src/lib.rs | 2 +
parquet/src/parquet_macros.rs | 264 +++++++++
parquet/src/parquet_thrift.rs | 415 +++++++++++++++
parquet/src/schema/printer.rs | 11 +-
parquet/tests/arrow_reader/bad_data.rs | 2 +-
6 files changed, 1243 insertions(+), 391 deletions(-)
diff --git a/parquet/src/basic.rs b/parquet/src/basic.rs
index de53c57ce5..92d70a933a 100644
--- a/parquet/src/basic.rs
+++ b/parquet/src/basic.rs
@@ -24,6 +24,8 @@ use std::str::FromStr;
use std::{fmt, str};
pub use crate::compression::{BrotliLevel, GzipLevel, ZstdLevel};
+use crate::parquet_thrift::{FieldType, ThriftCompactInputProtocol};
+use crate::{thrift_enum, thrift_private_struct, thrift_union_all_empty};
use crate::errors::{ParquetError, Result};
@@ -33,35 +35,29 @@ use crate::errors::{ParquetError, Result};
// ----------------------------------------------------------------------
// Mirrors thrift enum `crate::format::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.
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
-#[allow(non_camel_case_types)]
-pub enum Type {
- /// A boolean value.
- BOOLEAN,
- /// 32-bit signed integer.
- INT32,
- /// 64-bit signed integer.
- INT64,
- /// 96-bit signed integer for timestamps.
- INT96,
- /// IEEE 754 single-precision floating point value.
- FLOAT,
- /// IEEE 754 double-precision floating point value.
- DOUBLE,
- /// Arbitrary length byte array.
- BYTE_ARRAY,
- /// Fixed length byte array.
- FIXED_LEN_BYTE_ARRAY,
+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 `crate::format::ConvertedType`
+//
+// Cannot use macros because of added field `None`
/// Common types (converted types) used by frameworks when using Parquet.
///
@@ -165,23 +161,151 @@ pub enum ConvertedType {
INTERVAL,
}
+impl<'a> TryFrom<&mut ThriftCompactInputProtocol<'a>> for ConvertedType {
+ type Error = ParquetError;
+ fn try_from(prot: &mut ThriftCompactInputProtocol<'a>) -> Result<Self> {
+ let val = prot.read_i32()?;
+ Ok(match val {
+ 0 => Self::UTF8,
+ 1 => Self::MAP,
+ 2 => Self::MAP_KEY_VALUE,
+ 3 => Self::LIST,
+ 4 => Self::ENUM,
+ 5 => Self::DECIMAL,
+ 6 => Self::DATE,
+ 7 => Self::TIME_MILLIS,
+ 8 => Self::TIME_MICROS,
+ 9 => Self::TIMESTAMP_MILLIS,
+ 10 => Self::TIMESTAMP_MICROS,
+ 11 => Self::UINT_8,
+ 12 => Self::UINT_16,
+ 13 => Self::UINT_32,
+ 14 => Self::UINT_64,
+ 15 => Self::INT_8,
+ 16 => Self::INT_16,
+ 17 => Self::INT_32,
+ 18 => Self::INT_64,
+ 19 => Self::JSON,
+ 20 => Self::BSON,
+ 21 => Self::INTERVAL,
+ _ => return Err(general_err!("Unexpected ConvertedType {}", val)),
+ })
+ }
+}
+
// ----------------------------------------------------------------------
// Mirrors thrift union `crate::format::TimeUnit`
+thrift_union_all_empty!(
/// Time unit for `Time` and `Timestamp` logical types.
-#[derive(Clone, Debug, Eq, PartialEq)]
-pub enum TimeUnit {
- /// Milliseconds.
- MILLIS,
- /// Microseconds.
- MICROS,
- /// Nanoseconds.
- NANOS,
+union TimeUnit {
+ 1: MilliSeconds MILLIS
+ 2: MicroSeconds MICROS
+ 3: NanoSeconds NANOS
}
+);
// ----------------------------------------------------------------------
// Mirrors thrift union `crate::format::LogicalType`
+// private structs for decoding logical type
+
+thrift_private_struct!(
+struct DecimalType {
+ 1: required i32 scale
+ 2: required i32 precision
+}
+);
+
+thrift_private_struct!(
+struct TimestampType {
+ 1: required bool is_adjusted_to_u_t_c
+ 2: required TimeUnit unit
+}
+);
+
+// they are identical
+use TimestampType as TimeType;
+
+thrift_private_struct!(
+struct IntType {
+ 1: required i8 bit_width
+ 2: required bool is_signed
+}
+);
+
+thrift_private_struct!(
+struct VariantType {
+ // The version of the variant specification that the variant was
+ // written with.
+ 1: optional i8 specification_version
+}
+);
+
+// TODO need macro for structs that need lifetime annotation
+struct GeometryType<'a> {
+ crs: Option<&'a str>,
+}
+
+impl<'a> TryFrom<&mut ThriftCompactInputProtocol<'a>> for GeometryType<'a> {
+ type Error = ParquetError;
+ fn try_from(prot: &mut ThriftCompactInputProtocol<'a>) -> Result<Self> {
+ let mut crs: Option<&str> = None;
+ prot.read_struct_begin()?;
+ loop {
+ let field_ident = prot.read_field_begin()?;
+ if field_ident.field_type == FieldType::Stop {
+ break;
+ }
+ match field_ident.id {
+ 1 => {
+ let val = prot.read_string()?;
+ crs = Some(val);
+ }
+ _ => {
+ prot.skip(field_ident.field_type)?;
+ }
+ };
+ }
+ Ok(Self { crs })
+ }
+}
+
+struct GeographyType<'a> {
+ crs: Option<&'a str>,
+ algorithm: Option<EdgeInterpolationAlgorithm>,
+}
+
+impl<'a> TryFrom<&mut ThriftCompactInputProtocol<'a>> for GeographyType<'a> {
+ type Error = ParquetError;
+ fn try_from(prot: &mut ThriftCompactInputProtocol<'a>) -> Result<Self> {
+ let mut crs: Option<&str> = None;
+ let mut algorithm: Option<EdgeInterpolationAlgorithm> = None;
+ prot.read_struct_begin()?;
+ loop {
+ let field_ident = prot.read_field_begin()?;
+ if field_ident.field_type == FieldType::Stop {
+ break;
+ }
+ match field_ident.id {
+ 1 => {
+ let val = prot.read_string()?;
+ crs = Some(val);
+ }
+ 2 => {
+ let val = EdgeInterpolationAlgorithm::try_from(&mut *prot)?;
+ algorithm = Some(val);
+ }
+
+ _ => {
+ prot.skip(field_ident.field_type)?;
+ }
+ };
+ }
+ Ok(Self { crs, algorithm })
+ }
+}
+
/// Logical types used by version 2.4.0+ of the Parquet format.
///
/// This is an *entirely new* struct as of version
@@ -238,15 +362,149 @@ pub enum LogicalType {
/// A 16-bit floating point number.
Float16,
/// A Variant value.
- Variant,
+ Variant {
+ /// The version of the variant specification that the variant was written with.
+ specification_version: Option<i8>,
+ },
/// A geospatial feature in the Well-Known Binary (WKB) format with linear/planar edges interpolation.
- Geometry,
+ Geometry {
+ /// A custom CRS. If unset the defaults to `OGC:CRS84`.
+ crs: Option<String>,
+ },
/// A geospatial feature in the WKB format with an explicit (non-linear/non-planar) edges interpolation.
- Geography,
+ Geography {
+ /// A custom CRS. If unset the defaults to `OGC:CRS84`.
+ crs: Option<String>,
+ /// An optional algorithm can be set to correctly interpret edges interpolation
+ /// of the geometries. If unset, the algorithm defaults to `SPHERICAL``.
+ algorithm: Option<EdgeInterpolationAlgorithm>,
+ },
+ /// For forward compatibility; used when an unknown union value is encountered.
+ _Unknown {
+ /// The field id encountered when parsing the unknown logical type.
+ field_id: i16,
+ },
+}
+
+impl<'a> TryFrom<&mut ThriftCompactInputProtocol<'a>> for LogicalType {
+ type Error = ParquetError;
+ fn try_from(prot: &mut ThriftCompactInputProtocol<'a>) -> Result<Self> {
+ prot.read_struct_begin()?;
+
+ let field_ident = prot.read_field_begin()?;
+ if field_ident.field_type == FieldType::Stop {
+ return Err(general_err!("received empty union from remote LogicalType"));
+ }
+ let ret = match field_ident.id {
+ 1 => {
+ prot.skip_empty_struct()?;
+ Self::String
+ }
+ 2 => {
+ prot.skip_empty_struct()?;
+ Self::Map
+ }
+ 3 => {
+ prot.skip_empty_struct()?;
+ Self::List
+ }
+ 4 => {
+ prot.skip_empty_struct()?;
+ Self::Enum
+ }
+ 5 => {
+ let val = DecimalType::try_from(&mut *prot)?;
+ Self::Decimal {
+ scale: val.scale,
+ precision: val.precision,
+ }
+ }
+ 6 => {
+ prot.skip_empty_struct()?;
+ Self::Date
+ }
+ 7 => {
+ let val = TimeType::try_from(&mut *prot)?;
+ Self::Time {
+ is_adjusted_to_u_t_c: val.is_adjusted_to_u_t_c,
+ unit: val.unit,
+ }
+ }
+ 8 => {
+ let val = TimestampType::try_from(&mut *prot)?;
+ Self::Timestamp {
+ is_adjusted_to_u_t_c: val.is_adjusted_to_u_t_c,
+ unit: val.unit,
+ }
+ }
+ 10 => {
+ let val = IntType::try_from(&mut *prot)?;
+ Self::Integer {
+ is_signed: val.is_signed,
+ bit_width: val.bit_width,
+ }
+ }
+ 11 => {
+ prot.skip_empty_struct()?;
+ Self::Unknown
+ }
+ 12 => {
+ prot.skip_empty_struct()?;
+ Self::Json
+ }
+ 13 => {
+ prot.skip_empty_struct()?;
+ Self::Bson
+ }
+ 14 => {
+ prot.skip_empty_struct()?;
+ Self::Uuid
+ }
+ 15 => {
+ prot.skip_empty_struct()?;
+ Self::Float16
+ }
+ 16 => {
+ let val = VariantType::try_from(&mut *prot)?;
+ Self::Variant {
+ specification_version: val.specification_version,
+ }
+ }
+ 17 => {
+ let val = GeometryType::try_from(&mut *prot)?;
+ Self::Geometry {
+ crs: val.crs.map(|s| s.to_owned()),
+ }
+ }
+ 18 => {
+ let val = GeographyType::try_from(&mut *prot)?;
+ Self::Geography {
+ crs: val.crs.map(|s| s.to_owned()),
+ algorithm: val.algorithm,
+ }
+ }
+ _ => {
+ prot.skip(field_ident.field_type)?;
+ Self::_Unknown {
+ field_id: field_ident.id,
+ }
+ }
+ };
+ let field_ident = prot.read_field_begin()?;
+ if field_ident.field_type != FieldType::Stop {
+ return Err(general_err!(
+ "Received multiple fields for union from remote LogicalType"
+ ));
+ }
+ prot.read_struct_end()?;
+ Ok(ret)
+ }
}
// ----------------------------------------------------------------------
// Mirrors thrift enum `crate::format::FieldRepetitionType`
+//
+// Cannot use macro since the name is changed
/// Representation of field types in schema.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -260,9 +518,23 @@ pub enum Repetition {
REPEATED,
}
+impl<'a> TryFrom<&mut ThriftCompactInputProtocol<'a>> for Repetition {
+ type Error = ParquetError;
+ fn try_from(prot: &mut ThriftCompactInputProtocol<'a>) -> Result<Self> {
+ let val = prot.read_i32()?;
+ Ok(match val {
+ 0 => Self::REQUIRED,
+ 1 => Self::OPTIONAL,
+ 2 => Self::REPEATED,
+ _ => return Err(general_err!("Unexpected FieldRepetitionType {}", val)),
+ })
+ }
+}
+
// ----------------------------------------------------------------------
// Mirrors thrift enum `crate::format::Encoding`
+thrift_enum!(
/// Encodings supported by Parquet.
///
/// Not all encodings are valid for all types. These enums are also used to specify the
@@ -279,80 +551,19 @@ pub enum Repetition {
/// performance impact when evaluating these encodings.
///
/// [WriterVersion]: crate::file::properties::WriterVersion
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Ord, PartialOrd)]
-#[allow(non_camel_case_types)]
-pub enum Encoding {
- /// Default byte 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, stored as little-endian.
- /// - DOUBLE - 8 bytes per value, stored as little-endian.
- /// - BYTE_ARRAY - 4 byte length stored as little endian, followed by bytes.
- /// - FIXED_LEN_BYTE_ARRAY - just the bytes are stored.
- PLAIN,
-
- /// **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,
-
- /// Group packed run length encoding.
- ///
- /// Usable for definition/repetition levels encoding and boolean values.
- RLE,
-
- /// **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,
-
- /// Delta encoding for integers, either INT32 or INT64.
- ///
- /// Works best on sorted data.
- DELTA_BINARY_PACKED,
-
- /// 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,
-
- /// 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,
-
- /// Dictionary encoding.
- ///
- /// The ids are encoded using the RLE encoding.
- RLE_DICTIONARY,
-
- /// 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,
+enum Encoding {
+ PLAIN = 0;
+ // GROUP_VAR_INT = 1;
+ PLAIN_DICTIONARY = 2;
+ RLE = 3;
+ BIT_PACKED = 4;
+ DELTA_BINARY_PACKED = 5;
+ DELTA_LENGTH_BYTE_ARRAY = 6;
+ DELTA_BYTE_ARRAY = 7;
+ RLE_DICTIONARY = 8;
+ BYTE_STREAM_SPLIT = 9;
}
+);
impl FromStr for Encoding {
type Err = ParquetError;
@@ -415,6 +626,24 @@ pub enum Compression {
LZ4_RAW,
}
+impl<'a> TryFrom<&mut ThriftCompactInputProtocol<'a>> for Compression {
+ type Error = ParquetError;
+ fn try_from(prot: &mut ThriftCompactInputProtocol<'a>) -> Result<Self> {
+ let val = prot.read_i32()?;
+ Ok(match val {
+ 0 => Self::UNCOMPRESSED,
+ 1 => Self::SNAPPY,
+ 2 => Self::GZIP(Default::default()),
+ 3 => Self::LZO,
+ 4 => Self::BROTLI(Default::default()),
+ 5 => Self::LZ4,
+ 6 => Self::ZSTD(Default::default()),
+ 7 => Self::LZ4_RAW,
+ _ => return Err(general_err!("Unexpected CompressionCodec {}", val)),
+ })
+ }
+}
+
impl Compression {
/// Returns the codec type of this compression setting as a string, without the compression
/// level.
@@ -506,69 +735,78 @@ impl FromStr for Compression {
}
// ----------------------------------------------------------------------
-/// Mirrors thrift enum `crate::format::PageType`
-///
+// Mirrors thrift enum `crate::format::PageType`
+
+thrift_enum!(
/// Available data pages for Parquet file format.
/// Note that some of the page types may not be supported.
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-#[allow(non_camel_case_types)]
-pub enum PageType {
- /// Data page Parquet 1.0
- DATA_PAGE,
- /// Index page
- INDEX_PAGE,
- /// Dictionary page
- DICTIONARY_PAGE,
- /// Data page Parquet 2.0
- DATA_PAGE_V2,
+enum PageType {
+ DATA_PAGE = 0;
+ INDEX_PAGE = 1;
+ DICTIONARY_PAGE = 2;
+ DATA_PAGE_V2 = 3;
}
+);
// ----------------------------------------------------------------------
// Mirrors thrift enum `crate::format::BoundaryOrder`
+thrift_enum!(
/// Enum to annotate whether lists of min/max elements inside ColumnIndex
/// are ordered and if so, in which direction.
-#[derive(Debug, Clone, Copy, Hash, PartialEq, Eq)]
-pub enum BoundaryOrder {
- /// Min/max stats are unordered.
- UNORDERED,
- /// Min/max stats are ordered in an ascending fashion.
- ASCENDING,
- /// Min/max stats are ordered in an descending fashion.
- DESCENDING,
+enum BoundaryOrder {
+ UNORDERED = 0;
+ ASCENDING = 1;
+ DESCENDING = 2;
+}
+);
+
+// ----------------------------------------------------------------------
+// Mirrors thrift enum `crate::format::EdgeInterpolationAlgorithm`
+
+thrift_enum!(
+/// Edge interpolation algorithm for Geography logical type
+enum EdgeInterpolationAlgorithm {
+ SPHERICAL = 0;
+ VINCENTY = 1;
+ THOMAS = 2;
+ ANDOYER = 3;
+ KARNEY = 4;
}
+);
// ----------------------------------------------------------------------
// Mirrors thrift union `crate::format::BloomFilterAlgorithm`
+thrift_union_all_empty!(
/// The algorithm used in Bloom filter.
-#[derive(Clone, Debug, Eq, PartialEq)]
-pub enum BloomFilterAlgorithm {
- /// Block-based Bloom filter.
- BLOCK,
+union BloomFilterAlgorithm {
+ /** Block-based Bloom filter. **/
+ 1: SplitBlockAlgorithm BLOCK;
}
+);
// ----------------------------------------------------------------------
// Mirrors thrift union `crate::format::BloomFilterHash`
+thrift_union_all_empty!(
/// The hash function used in Bloom filter. This function takes the hash of a column value
/// using plain encoding.
-#[derive(Clone, Debug, Eq, PartialEq)]
-pub enum BloomFilterHash {
- /// xxHash is an extremely fast non-cryptographic hash algorithm. It uses 64 bits version
- /// of xxHash.
- XXHASH,
+union BloomFilterHash {
+ /** xxHash Strategy. **/
+ 1: XxHash XXHASH;
}
+);
// ----------------------------------------------------------------------
// Mirrors thrift union `crate::format::BloomFilterCompression`
+thrift_union_all_empty!(
/// The compression used in the Bloom filter.
-#[derive(Clone, Debug, Eq, PartialEq)]
-pub enum BloomFilterCompression {
- /// No compression is used.
- UNCOMPRESSED,
+union BloomFilterCompression {
+ 1: Uncompressed UNCOMPRESSED;
}
+);
// ----------------------------------------------------------------------
// Mirrors thrift union `crate::format::ColumnOrder`
@@ -610,9 +848,13 @@ 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),
+ // 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 {
@@ -640,9 +882,10 @@ impl ColumnOrder {
LogicalType::Unknown => SortOrder::UNDEFINED,
LogicalType::Uuid => SortOrder::UNSIGNED,
LogicalType::Float16 => SortOrder::SIGNED,
- LogicalType::Variant | LogicalType::Geometry | LogicalType::Geography => {
- SortOrder::UNDEFINED
- }
+ 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),
@@ -712,19 +955,45 @@ impl ColumnOrder {
match *self {
ColumnOrder::TYPE_DEFINED_ORDER(order) => order,
ColumnOrder::UNDEFINED => SortOrder::SIGNED,
+ ColumnOrder::UNKNOWN => SortOrder::UNDEFINED,
}
}
}
-// ----------------------------------------------------------------------
-// Display handlers
+impl<'a> TryFrom<&mut ThriftCompactInputProtocol<'a>> for ColumnOrder {
+ type Error = ParquetError;
-impl fmt::Display for Type {
- fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
- write!(f, "{self:?}")
+ fn try_from(prot: &mut ThriftCompactInputProtocol<'a>) -> Result<Self> {
+ prot.read_struct_begin()?;
+ let field_ident = prot.read_field_begin()?;
+ if field_ident.field_type == FieldType::Stop {
+ return Err(general_err!("Received empty union from remote ColumnOrder"));
+ }
+ let ret = match field_ident.id {
+ 1 => {
+ // TODO: the sort order needs to be set correctly after parsing.
+ prot.skip_empty_struct()?;
+ Self::TYPE_DEFINED_ORDER(SortOrder::SIGNED)
+ }
+ _ => {
+ prot.skip(field_ident.field_type)?;
+ Self::UNKNOWN
+ }
+ };
+ let field_ident = prot.read_field_begin()?;
+ if field_ident.field_type != FieldType::Stop {
+ return Err(general_err!(
+ "Received multiple fields for union from remote ColumnOrder"
+ ));
+ }
+ prot.read_struct_end()?;
+ Ok(ret)
}
}
+// ----------------------------------------------------------------------
+// Display handlers
+
impl fmt::Display for ConvertedType {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
write!(f, "{self:?}")
@@ -737,24 +1006,12 @@ impl fmt::Display for Repetition {
}
}
-impl fmt::Display for Encoding {
- fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
- write!(f, "{self:?}")
- }
-}
-
impl fmt::Display for Compression {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
write!(f, "{self:?}")
}
}
-impl fmt::Display for PageType {
- 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:?}")
@@ -767,42 +1024,6 @@ impl fmt::Display for ColumnOrder {
}
}
-// ----------------------------------------------------------------------
-// crate::format::Type <=> Type conversion
-
-impl TryFrom<crate::format::Type> for Type {
- type Error = ParquetError;
-
- fn try_from(value: crate::format::Type) -> Result<Self> {
- Ok(match value {
- crate::format::Type::BOOLEAN => Type::BOOLEAN,
- crate::format::Type::INT32 => Type::INT32,
- crate::format::Type::INT64 => Type::INT64,
- crate::format::Type::INT96 => Type::INT96,
- crate::format::Type::FLOAT => Type::FLOAT,
- crate::format::Type::DOUBLE => Type::DOUBLE,
- crate::format::Type::BYTE_ARRAY => Type::BYTE_ARRAY,
- crate::format::Type::FIXED_LEN_BYTE_ARRAY => Type::FIXED_LEN_BYTE_ARRAY,
- _ => return Err(general_err!("unexpected parquet type: {}", value.0)),
- })
- }
-}
-
-impl From<Type> for crate::format::Type {
- fn from(value: Type) -> Self {
- match value {
- Type::BOOLEAN => crate::format::Type::BOOLEAN,
- Type::INT32 => crate::format::Type::INT32,
- Type::INT64 => crate::format::Type::INT64,
- Type::INT96 => crate::format::Type::INT96,
- Type::FLOAT => crate::format::Type::FLOAT,
- Type::DOUBLE => crate::format::Type::DOUBLE,
- Type::BYTE_ARRAY => crate::format::Type::BYTE_ARRAY,
- Type::FIXED_LEN_BYTE_ARRAY => crate::format::Type::FIXED_LEN_BYTE_ARRAY,
- }
- }
-}
-
// ----------------------------------------------------------------------
// crate::format::ConvertedType <=> ConvertedType conversion
@@ -876,92 +1097,6 @@ impl From<ConvertedType> for Option<crate::format::ConvertedType> {
}
}
-// ----------------------------------------------------------------------
-// crate::format::BloomFilterHash <=> BloomFilterHash conversion
-
-impl From<crate::format::BloomFilterHash> for BloomFilterHash {
- fn from(value: crate::format::BloomFilterHash) -> Self {
- match value {
- crate::format::BloomFilterHash::XXHASH(_) => BloomFilterHash::XXHASH,
- }
- }
-}
-
-impl From<BloomFilterHash> for crate::format::BloomFilterHash {
- fn from(value: BloomFilterHash) -> Self {
- match value {
- BloomFilterHash::XXHASH => crate::format::BloomFilterHash::XXHASH(Default::default()),
- }
- }
-}
-
-// ----------------------------------------------------------------------
-// crate::format::BloomFilterAlgorithm <=> BloomFilterAlgorithm conversion
-
-impl From<crate::format::BloomFilterAlgorithm> for BloomFilterAlgorithm {
- fn from(value: crate::format::BloomFilterAlgorithm) -> Self {
- match value {
- crate::format::BloomFilterAlgorithm::BLOCK(_) => BloomFilterAlgorithm::BLOCK,
- }
- }
-}
-
-impl From<BloomFilterAlgorithm> for crate::format::BloomFilterAlgorithm {
- fn from(value: BloomFilterAlgorithm) -> Self {
- match value {
- BloomFilterAlgorithm::BLOCK => {
- crate::format::BloomFilterAlgorithm::BLOCK(Default::default())
- }
- }
- }
-}
-
-// ----------------------------------------------------------------------
-// crate::format::BloomFilterCompression <=> BloomFilterCompression conversion
-
-impl From<crate::format::BloomFilterCompression> for BloomFilterCompression {
- fn from(value: crate::format::BloomFilterCompression) -> Self {
- match value {
- crate::format::BloomFilterCompression::UNCOMPRESSED(_) => {
- BloomFilterCompression::UNCOMPRESSED
- }
- }
- }
-}
-
-impl From<BloomFilterCompression> for crate::format::BloomFilterCompression {
- fn from(value: BloomFilterCompression) -> Self {
- match value {
- BloomFilterCompression::UNCOMPRESSED => {
- crate::format::BloomFilterCompression::UNCOMPRESSED(Default::default())
- }
- }
- }
-}
-
-// ----------------------------------------------------------------------
-// crate::format::TimeUnit <=> TimeUnit conversion
-
-impl From<crate::format::TimeUnit> for TimeUnit {
- fn from(value: crate::format::TimeUnit) -> Self {
- match value {
- crate::format::TimeUnit::MILLIS(_) => TimeUnit::MILLIS,
- crate::format::TimeUnit::MICROS(_) => TimeUnit::MICROS,
- crate::format::TimeUnit::NANOS(_) => TimeUnit::NANOS,
- }
- }
-}
-
-impl From<TimeUnit> for crate::format::TimeUnit {
- fn from(value: TimeUnit) -> Self {
- match value {
- TimeUnit::MILLIS => crate::format::TimeUnit::MILLIS(crate::format::MilliSeconds {}),
- TimeUnit::MICROS => crate::format::TimeUnit::MICROS(crate::format::MicroSeconds {}),
- TimeUnit::NANOS => crate::format::TimeUnit::NANOS(crate::format::NanoSeconds {}),
- }
- }
-}
-
// ----------------------------------------------------------------------
// crate::format::LogicalType <=> LogicalType conversion
@@ -994,9 +1129,14 @@ impl From<crate::format::LogicalType> for LogicalType {
crate::format::LogicalType::BSON(_) => LogicalType::Bson,
crate::format::LogicalType::UUID(_) => LogicalType::Uuid,
crate::format::LogicalType::FLOAT16(_) => LogicalType::Float16,
- crate::format::LogicalType::VARIANT(_) => LogicalType::Variant,
- crate::format::LogicalType::GEOMETRY(_) => LogicalType::Geometry,
- crate::format::LogicalType::GEOGRAPHY(_) => LogicalType::Geography,
+ crate::format::LogicalType::VARIANT(vt) => LogicalType::Variant {
+ specification_version: vt.specification_version,
+ },
+ crate::format::LogicalType::GEOMETRY(gt) => LogicalType::Geometry { crs: gt.crs },
+ crate::format::LogicalType::GEOGRAPHY(gt) => LogicalType::Geography {
+ crs: gt.crs,
+ algorithm: gt.algorithm.map(|a| a.try_into().unwrap()),
+ },
}
}
}
@@ -1038,9 +1178,23 @@ impl From<LogicalType> for crate::format::LogicalType {
LogicalType::Bson => crate::format::LogicalType::BSON(Default::default()),
LogicalType::Uuid => crate::format::LogicalType::UUID(Default::default()),
LogicalType::Float16 => crate::format::LogicalType::FLOAT16(Default::default()),
- LogicalType::Variant => crate::format::LogicalType::VARIANT(Default::default()),
- LogicalType::Geometry => crate::format::LogicalType::GEOMETRY(Default::default()),
- LogicalType::Geography => crate::format::LogicalType::GEOGRAPHY(Default::default()),
+ LogicalType::Variant {
+ specification_version,
+ } => crate::format::LogicalType::VARIANT(crate::format::VariantType {
+ specification_version,
+ }),
+ LogicalType::Geometry { crs } => {
+ crate::format::LogicalType::GEOMETRY(crate::format::GeometryType { crs })
+ }
+ LogicalType::Geography { crs, algorithm } => {
+ crate::format::LogicalType::GEOGRAPHY(crate::format::GeographyType {
+ crs,
+ algorithm: algorithm.map(|a| a.into()),
+ })
+ }
+ LogicalType::_Unknown { .. } => {
+ panic!("Trying to convert unknown LogicalType to thrift");
+ }
}
}
}
@@ -1092,9 +1246,10 @@ impl From<Option<LogicalType>> for ConvertedType {
LogicalType::Bson => ConvertedType::BSON,
LogicalType::Uuid
| LogicalType::Float16
- | LogicalType::Variant
- | LogicalType::Geometry
- | LogicalType::Geography
+ | LogicalType::Variant { .. }
+ | LogicalType::Geometry { .. }
+ | LogicalType::Geography { .. }
+ | LogicalType::_Unknown { .. }
| LogicalType::Unknown => ConvertedType::NONE,
},
None => ConvertedType::NONE,
@@ -1133,46 +1288,6 @@ impl From<Repetition> for crate::format::FieldRepetitionType {
}
}
-// ----------------------------------------------------------------------
-// crate::format::Encoding <=> Encoding conversion
-
-impl TryFrom<crate::format::Encoding> for Encoding {
- type Error = ParquetError;
-
- fn try_from(value: crate::format::Encoding) -> Result<Self> {
- Ok(match value {
- crate::format::Encoding::PLAIN => Encoding::PLAIN,
- crate::format::Encoding::PLAIN_DICTIONARY => Encoding::PLAIN_DICTIONARY,
- crate::format::Encoding::RLE => Encoding::RLE,
- #[allow(deprecated)]
- crate::format::Encoding::BIT_PACKED => Encoding::BIT_PACKED,
- crate::format::Encoding::DELTA_BINARY_PACKED => Encoding::DELTA_BINARY_PACKED,
- crate::format::Encoding::DELTA_LENGTH_BYTE_ARRAY => Encoding::DELTA_LENGTH_BYTE_ARRAY,
- crate::format::Encoding::DELTA_BYTE_ARRAY => Encoding::DELTA_BYTE_ARRAY,
- crate::format::Encoding::RLE_DICTIONARY => Encoding::RLE_DICTIONARY,
- crate::format::Encoding::BYTE_STREAM_SPLIT => Encoding::BYTE_STREAM_SPLIT,
- _ => return Err(general_err!("unexpected parquet encoding: {}", value.0)),
- })
- }
-}
-
-impl From<Encoding> for crate::format::Encoding {
- fn from(value: Encoding) -> Self {
- match value {
- Encoding::PLAIN => crate::format::Encoding::PLAIN,
- Encoding::PLAIN_DICTIONARY => crate::format::Encoding::PLAIN_DICTIONARY,
- Encoding::RLE => crate::format::Encoding::RLE,
- #[allow(deprecated)]
- Encoding::BIT_PACKED => crate::format::Encoding::BIT_PACKED,
- Encoding::DELTA_BINARY_PACKED => crate::format::Encoding::DELTA_BINARY_PACKED,
- Encoding::DELTA_LENGTH_BYTE_ARRAY => crate::format::Encoding::DELTA_LENGTH_BYTE_ARRAY,
- Encoding::DELTA_BYTE_ARRAY => crate::format::Encoding::DELTA_BYTE_ARRAY,
- Encoding::RLE_DICTIONARY => crate::format::Encoding::RLE_DICTIONARY,
- Encoding::BYTE_STREAM_SPLIT => crate::format::Encoding::BYTE_STREAM_SPLIT,
- }
- }
-}
-
// ----------------------------------------------------------------------
// crate::format::CompressionCodec <=> Compression conversion
@@ -1214,65 +1329,6 @@ impl From<Compression> for crate::format::CompressionCodec {
}
}
-// ----------------------------------------------------------------------
-// crate::format::PageType <=> PageType conversion
-
-impl TryFrom<crate::format::PageType> for PageType {
- type Error = ParquetError;
-
- fn try_from(value: crate::format::PageType) -> Result<Self> {
- Ok(match value {
- crate::format::PageType::DATA_PAGE => PageType::DATA_PAGE,
- crate::format::PageType::INDEX_PAGE => PageType::INDEX_PAGE,
- crate::format::PageType::DICTIONARY_PAGE => PageType::DICTIONARY_PAGE,
- crate::format::PageType::DATA_PAGE_V2 => PageType::DATA_PAGE_V2,
- _ => return Err(general_err!("unexpected parquet page type: {}", value.0)),
- })
- }
-}
-
-impl From<PageType> for crate::format::PageType {
- fn from(value: PageType) -> Self {
- match value {
- PageType::DATA_PAGE => crate::format::PageType::DATA_PAGE,
- PageType::INDEX_PAGE => crate::format::PageType::INDEX_PAGE,
- PageType::DICTIONARY_PAGE => crate::format::PageType::DICTIONARY_PAGE,
- PageType::DATA_PAGE_V2 => crate::format::PageType::DATA_PAGE_V2,
- }
- }
-}
-
-// ----------------------------------------------------------------------
-// crate::format::PageType <=> PageType conversion
-
-impl TryFrom<crate::format::BoundaryOrder> for BoundaryOrder {
- type Error = ParquetError;
-
- fn try_from(value: crate::format::BoundaryOrder) -> Result<Self> {
- Ok(match value {
- crate::format::BoundaryOrder::UNORDERED => BoundaryOrder::UNORDERED,
- crate::format::BoundaryOrder::ASCENDING => BoundaryOrder::ASCENDING,
- crate::format::BoundaryOrder::DESCENDING => BoundaryOrder::DESCENDING,
- _ => {
- return Err(general_err!(
- "unexpected parquet boundary order type: {}",
- value.0
- ))
- }
- })
- }
-}
-
-impl From<BoundaryOrder> for crate::format::BoundaryOrder {
- fn from(value: BoundaryOrder) -> Self {
- match value {
- BoundaryOrder::UNORDERED => crate::format::BoundaryOrder::UNORDERED,
- BoundaryOrder::ASCENDING => crate::format::BoundaryOrder::ASCENDING,
- BoundaryOrder::DESCENDING => crate::format::BoundaryOrder::DESCENDING,
- }
- }
-}
-
// ----------------------------------------------------------------------
// String conversions for schema parsing.
@@ -2633,4 +2689,114 @@ mod tests {
"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_from_boundary_order() {
+ assert_eq!(
+ BoundaryOrder::try_from(crate::format::BoundaryOrder::ASCENDING).unwrap(),
+ BoundaryOrder::ASCENDING
+ );
+ assert_eq!(
+ BoundaryOrder::try_from(crate::format::BoundaryOrder::DESCENDING).unwrap(),
+ BoundaryOrder::DESCENDING
+ );
+ assert_eq!(
+ BoundaryOrder::try_from(crate::format::BoundaryOrder::UNORDERED).unwrap(),
+ BoundaryOrder::UNORDERED
+ );
+ }
+
+ #[test]
+ fn test_into_boundary_order() {
+ assert_eq!(
+ crate::format::BoundaryOrder::ASCENDING,
+ BoundaryOrder::ASCENDING.into()
+ );
+ assert_eq!(
+ crate::format::BoundaryOrder::DESCENDING,
+ BoundaryOrder::DESCENDING.into()
+ );
+ assert_eq!(
+ crate::format::BoundaryOrder::UNORDERED,
+ BoundaryOrder::UNORDERED.into()
+ );
+ }
+
+ #[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_edge_algo() {
+ assert_eq!(
+ EdgeInterpolationAlgorithm::try_from(
+ crate::format::EdgeInterpolationAlgorithm::SPHERICAL
+ )
+ .unwrap(),
+ EdgeInterpolationAlgorithm::SPHERICAL
+ );
+ assert_eq!(
+ EdgeInterpolationAlgorithm::try_from(
+ crate::format::EdgeInterpolationAlgorithm::VINCENTY
+ )
+ .unwrap(),
+ EdgeInterpolationAlgorithm::VINCENTY
+ );
+ assert_eq!(
+ EdgeInterpolationAlgorithm::try_from(crate::format::EdgeInterpolationAlgorithm::THOMAS)
+ .unwrap(),
+ EdgeInterpolationAlgorithm::THOMAS
+ );
+ assert_eq!(
+ EdgeInterpolationAlgorithm::try_from(
+ crate::format::EdgeInterpolationAlgorithm::ANDOYER
+ )
+ .unwrap(),
+ EdgeInterpolationAlgorithm::ANDOYER
+ );
+ assert_eq!(
+ EdgeInterpolationAlgorithm::try_from(crate::format::EdgeInterpolationAlgorithm::KARNEY)
+ .unwrap(),
+ EdgeInterpolationAlgorithm::KARNEY
+ );
+ }
+
+ #[test]
+ fn test_into_edge_algo() {
+ assert_eq!(
+ crate::format::EdgeInterpolationAlgorithm::SPHERICAL,
+ EdgeInterpolationAlgorithm::SPHERICAL.into()
+ );
+ assert_eq!(
+ crate::format::EdgeInterpolationAlgorithm::VINCENTY,
+ EdgeInterpolationAlgorithm::VINCENTY.into()
+ );
+ assert_eq!(
+ crate::format::EdgeInterpolationAlgorithm::THOMAS,
+ EdgeInterpolationAlgorithm::THOMAS.into()
+ );
+ assert_eq!(
+ crate::format::EdgeInterpolationAlgorithm::ANDOYER,
+ EdgeInterpolationAlgorithm::ANDOYER.into()
+ );
+ assert_eq!(
+ crate::format::EdgeInterpolationAlgorithm::KARNEY,
+ EdgeInterpolationAlgorithm::KARNEY.into()
+ );
+ }
}
diff --git a/parquet/src/lib.rs b/parquet/src/lib.rs
index 07a673c295..74182d2650 100644
--- a/parquet/src/lib.rs
+++ b/parquet/src/lib.rs
@@ -178,4 +178,6 @@ pub mod file;
pub mod record;
pub mod schema;
+mod parquet_macros;
+mod parquet_thrift;
pub mod thrift;
diff --git a/parquet/src/parquet_macros.rs b/parquet/src/parquet_macros.rs
new file mode 100644
index 0000000000..40e148a75e
--- /dev/null
+++ b/parquet/src/parquet_macros.rs
@@ -0,0 +1,264 @@
+// 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.
+
+// These macros are adapted from Jörn Horstmann's thrift macros at
+// https://github.com/jhorstmann/compact-thrift
+// They allow for pasting sections of the Parquet thrift IDL file
+// into a macro to generate rust structures and implementations.
+
+#[macro_export]
+#[allow(clippy::crate_in_macro_def)]
+/// macro to generate rust enums from a thrift enum definition
+macro_rules! thrift_enum {
+ ($(#[$($def_attrs:tt)*])* enum $identifier:ident { $($(#[$($field_attrs:tt)*])* $field_name:ident = $field_value:literal;)* }) => {
+ $(#[$($def_attrs)*])*
+ #[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
+ #[allow(non_camel_case_types)]
+ #[allow(missing_docs)]
+ pub enum $identifier {
+ $($field_name = $field_value,)*
+ }
+
+ impl<'a> TryFrom<&mut ThriftCompactInputProtocol<'a>> for $identifier {
+ type Error = ParquetError;
+ fn try_from(prot: &mut ThriftCompactInputProtocol<'a>) -> Result<Self> {
+ let val = prot.read_i32()?;
+ match val {
+ $($field_value => Ok(Self::$field_name),)*
+ _ => Err(general_err!("Unexpected {} {}", stringify!($identifier), val)),
+ }
+ }
+ }
+
+ impl fmt::Display for $identifier {
+ fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+ write!(f, "{self:?}")
+ }
+ }
+
+ // TODO: remove when we finally get rid of the format module
+ impl TryFrom<crate::format::$identifier> for $identifier {
+ type Error = ParquetError;
+
+ fn try_from(value: crate::format::$identifier) -> Result<Self> {
+ Ok(match value {
+ $(crate::format::$identifier::$field_name => Self::$field_name,)*
+ _ => return Err(general_err!("Unexpected parquet {}: {}", stringify!($identifier), value.0)),
+ })
+ }
+ }
+
+ impl From<$identifier> for crate::format::$identifier {
+ fn from(value: $identifier) -> Self {
+ match value {
+ $($identifier::$field_name => Self::$field_name,)*
+ }
+ }
+ }
+ }
+}
+
+#[macro_export]
+#[allow(clippy::crate_in_macro_def)]
+/// macro to generate rust enums for empty thrift structs used in unions
+macro_rules! thrift_empty_struct {
+ ($identifier: ident) => {
+ #[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
+ pub struct $identifier {}
+
+ impl<'a> TryFrom<&mut ThriftCompactInputProtocol<'a>> for $identifier {
+ type Error = ParquetError;
+ fn try_from(prot: &mut ThriftCompactInputProtocol<'a>) -> Result<Self> {
+ prot.skip_empty_struct()?;
+ Ok(Self {})
+ }
+ }
+
+ // TODO: remove when we finally get rid of the format module
+ impl From<crate::format::$identifier> for $identifier {
+ fn from(_: $crate::format::$identifier) -> Self {
+ Self {}
+ }
+ }
+
+ impl From<$identifier> for crate::format::$identifier {
+ fn from(_: $identifier) -> Self {
+ Self {}
+ }
+ }
+ };
+}
+
+/// macro to generate rust enums for thrift unions where all fields are typed with empty structs
+#[macro_export]
+#[allow(clippy::crate_in_macro_def)]
+macro_rules! thrift_union_all_empty {
+ ($(#[$($def_attrs:tt)*])* union $identifier:ident { $($(#[$($field_attrs:tt)*])* $field_id:literal : $field_type:ident $(< $element_type:ident >)? $field_name:ident $(;)?)* }) => {
+ $(#[cfg_attr(not(doctest), $($def_attrs)*)])*
+ #[derive(Clone, Copy, Debug, Eq, PartialEq)]
+ #[allow(non_camel_case_types)]
+ #[allow(non_snake_case)]
+ #[allow(missing_docs)]
+ pub enum $identifier {
+ $($(#[cfg_attr(not(doctest), $($field_attrs)*)])* $field_name),*
+ }
+
+ impl<'a> TryFrom<&mut ThriftCompactInputProtocol<'a>> for $identifier {
+ type Error = ParquetError;
+
+ fn try_from(prot: &mut ThriftCompactInputProtocol<'a>) -> Result<Self> {
+ prot.read_struct_begin()?;
+ let field_ident = prot.read_field_begin()?;
+ if field_ident.field_type == FieldType::Stop {
+ return Err(general_err!("Received empty union from remote {}", stringify!($identifier)));
+ }
+ let ret = match field_ident.id {
+ $($field_id => {
+ prot.skip_empty_struct()?;
+ Self::$field_name
+ }
+ )*
+ _ => {
+ return Err(general_err!("Unexpected {} {}", stringify!($identifier), field_ident.id));
+ }
+ };
+ let field_ident = prot.read_field_begin()?;
+ if field_ident.field_type != FieldType::Stop {
+ return Err(general_err!(
+ "Received multiple fields for union from remote {}", stringify!($identifier)
+ ));
+ }
+ prot.read_struct_end()?;
+ Ok(ret)
+ }
+ }
+
+ // TODO: remove when we finally get rid of the format module
+ impl From<crate::format::$identifier> for $identifier {
+ fn from(value: crate::format::$identifier) -> Self {
+ match value {
+ $(crate::format::$identifier::$field_name(_) => Self::$field_name,)*
+ }
+ }
+ }
+
+ impl From<$identifier> for crate::format::$identifier {
+ fn from(value: $identifier) -> Self {
+ match value {
+ $($identifier::$field_name => Self::$field_name(Default::default()),)*
+ }
+ }
+ }
+ }
+}
+
+/// macro to generate rust structs from a thrift struct definition
+#[macro_export]
+macro_rules! thrift_private_struct {
+ ($(#[$($def_attrs:tt)*])* struct $identifier:ident { $($(#[$($field_attrs:tt)*])* $field_id:literal : $required_or_optional:ident $field_type:ident $(< $element_type:ident >)? $field_name:ident $(= $default_value:literal)? $(;)?)* }) => {
+ $(#[cfg_attr(not(doctest), $($def_attrs)*)])*
+ #[derive(Clone, Debug, PartialEq)]
+ #[allow(non_camel_case_types)]
+ #[allow(non_snake_case)]
+ struct $identifier {
+ $($(#[cfg_attr(not(doctest), $($field_attrs)*)])* pub $field_name: $crate::__thrift_required_or_optional!($required_or_optional $crate::__thrift_field_type!($field_type $($element_type)?))),*
+ }
+
+ impl<'a> TryFrom<&mut ThriftCompactInputProtocol<'a>> for $identifier {
+ type Error = ParquetError;
+ fn try_from(prot: &mut ThriftCompactInputProtocol<'a>) -> Result<Self> {
+ $(let mut $field_name: Option<$field_type> = None;)*
+ prot.read_struct_begin()?;
+ loop {
+ let field_ident = prot.read_field_begin()?;
+ if field_ident.field_type == FieldType::Stop {
+ break;
+ }
+ match field_ident.id {
+ $($field_id => {
+ let val = $crate::__thrift_read_field!(prot $field_type);
+ $field_name = Some(val);
+ })*
+ _ => {
+ prot.skip(field_ident.field_type)?;
+ }
+ };
+ }
+ Ok(Self {
+ $($field_name: $crate::__thrift_result_required_or_optional!($required_or_optional $field_name)),*
+ })
+ }
+ }
+ }
+}
+
+#[doc(hidden)]
+#[macro_export]
+macro_rules! __thrift_required_or_optional {
+ (required $field_type:ty) => { $field_type };
+ (optional $field_type:ty) => { Option<$field_type> };
+}
+
+#[doc(hidden)]
+#[macro_export]
+macro_rules! __thrift_result_required_or_optional {
+ (required $field_name:ident) => {
+ $field_name.expect(&format!(
+ "Required field {} not present",
+ stringify!($field_name)
+ ))
+ };
+ (optional $field_name:ident) => {
+ $field_name
+ };
+}
+
+#[doc(hidden)]
+#[macro_export]
+macro_rules! __thrift_read_field {
+ ($prot:tt bool) => {
+ $prot.read_bool()?
+ };
+ ($prot:tt i8) => {
+ $prot.read_i8()?
+ };
+ ($prot:tt i32) => {
+ $prot.read_i32()?
+ };
+ ($prot:tt i64) => {
+ $prot.read_i64()?
+ };
+ ($prot:tt string) => {
+ $prot.read_string()?
+ };
+ ($prot:tt $field_type:ident) => {
+ $field_type::try_from(&mut *$prot)?
+ };
+}
+
+#[doc(hidden)]
+#[macro_export]
+macro_rules! __thrift_field_type {
+ (list $element_type:ident) => { Vec< $crate::__thrift_field_type!($element_type) > };
+ (set $element_type:ident) => { Vec< $crate::__thrift_field_type!($element_type) > };
+ (binary) => { Vec<u8> };
+ (string) => { String };
+ ($field_type:ty) => { $field_type };
+ (Box $element_type:ident) => { std::boxed::Box< $crate::field_type!($element_type) > };
+ (Rc $element_type:ident) => { std::rc::Rc< $crate::__thrift_field_type!($element_type) > };
+ (Arc $element_type:ident) => { std::sync::Arc< $crate::__thrift_field_type!($element_type) > };
+}
diff --git a/parquet/src/parquet_thrift.rs b/parquet/src/parquet_thrift.rs
new file mode 100644
index 0000000000..2514f3fc70
--- /dev/null
+++ b/parquet/src/parquet_thrift.rs
@@ -0,0 +1,415 @@
+// 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.
+
+//! experimental replacement for thrift decoder
+// this is a copy of TCompactSliceInputProtocol, but modified
+// to not allocate byte arrays or strings.
+#![allow(dead_code)]
+
+use crate::errors::{ParquetError, Result};
+
+// Thrift compact protocol types for struct fields.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub(crate) enum FieldType {
+ Stop = 0,
+ BooleanTrue = 1,
+ BooleanFalse = 2,
+ Byte = 3,
+ I16 = 4,
+ I32 = 5,
+ I64 = 6,
+ Double = 7,
+ Binary = 8,
+ List = 9,
+ Set = 10,
+ Map = 11,
+ Struct = 12,
+}
+
+impl TryFrom<u8> for FieldType {
+ type Error = ParquetError;
+ fn try_from(value: u8) -> Result<Self> {
+ match value {
+ 0 => Ok(Self::Stop),
+ 1 => Ok(Self::BooleanTrue),
+ 2 => Ok(Self::BooleanFalse),
+ 3 => Ok(Self::Byte),
+ 4 => Ok(Self::I16),
+ 5 => Ok(Self::I32),
+ 6 => Ok(Self::I64),
+ 7 => Ok(Self::Double),
+ 8 => Ok(Self::Binary),
+ 9 => Ok(Self::List),
+ 10 => Ok(Self::Set),
+ 11 => Ok(Self::Map),
+ 12 => Ok(Self::Struct),
+ _ => Err(general_err!("Unexpected struct field type{}", value)),
+ }
+ }
+}
+
+impl TryFrom<ElementType> for FieldType {
+ type Error = ParquetError;
+ fn try_from(value: ElementType) -> std::result::Result<Self, Self::Error> {
+ match value {
+ ElementType::Bool => Ok(Self::BooleanTrue),
+ ElementType::Byte => Ok(Self::Byte),
+ ElementType::I16 => Ok(Self::I16),
+ ElementType::I32 => Ok(Self::I32),
+ ElementType::I64 => Ok(Self::I64),
+ ElementType::Double => Ok(Self::Double),
+ ElementType::Binary => Ok(Self::Binary),
+ ElementType::List => Ok(Self::List),
+ ElementType::Struct => Ok(Self::Struct),
+ _ => Err(general_err!("Unexpected list element type{:?}", value)),
+ }
+ }
+}
+
+// Thrift compact protocol types for list elements
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub(crate) enum ElementType {
+ Bool = 2,
+ Byte = 3,
+ I16 = 4,
+ I32 = 5,
+ I64 = 6,
+ Double = 7,
+ Binary = 8,
+ List = 9,
+ Set = 10,
+ Map = 11,
+ Struct = 12,
+}
+
+impl TryFrom<u8> for ElementType {
+ type Error = ParquetError;
+ fn try_from(value: u8) -> Result<Self> {
+ match value {
+ // For historical and compatibility reasons, a reader should be capable to deal with both cases.
+ // The only valid value in the original spec was 2, but due to an widespread implementation bug
+ // the defacto standard across large parts of the library became 1 instead.
+ // As a result, both values are now allowed.
+ // https://github.com/apache/thrift/blob/master/doc/specs/thrift-compact-protocol.md#list-and-set
+ 1 | 2 => Ok(Self::Bool),
+ 3 => Ok(Self::Byte),
+ 4 => Ok(Self::I16),
+ 5 => Ok(Self::I32),
+ 6 => Ok(Self::I64),
+ 7 => Ok(Self::Double),
+ 8 => Ok(Self::Binary),
+ 9 => Ok(Self::List),
+ 10 => Ok(Self::Set),
+ 11 => Ok(Self::Map),
+ 12 => Ok(Self::Struct),
+ _ => Err(general_err!("Unexpected list/set element type{}", value)),
+ }
+ }
+}
+
+pub(crate) struct FieldIdentifier {
+ pub(crate) field_type: FieldType,
+ pub(crate) id: i16,
+}
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub(crate) struct ListIdentifier {
+ pub(crate) element_type: ElementType,
+ pub(crate) size: i32,
+}
+
+/// A more performant implementation of [`TCompactInputProtocol`] that reads a slice
+///
+/// [`TCompactInputProtocol`]: thrift::protocol::TCompactInputProtocol
+pub(crate) struct ThriftCompactInputProtocol<'a> {
+ buf: &'a [u8],
+ // Identifier of the last field deserialized for a struct.
+ last_read_field_id: i16,
+ // Stack of the last read field ids (a new entry is added each time a nested struct is read).
+ read_field_id_stack: Vec<i16>,
+ // Boolean value for a field.
+ // Saved because boolean fields and their value are encoded in a single byte,
+ // and reading the field only occurs after the field id is read.
+ pending_read_bool_value: Option<bool>,
+}
+
+impl<'b, 'a: 'b> ThriftCompactInputProtocol<'a> {
+ pub fn new(buf: &'a [u8]) -> Self {
+ Self {
+ buf,
+ last_read_field_id: 0,
+ read_field_id_stack: Vec::with_capacity(16),
+ pending_read_bool_value: None,
+ }
+ }
+
+ pub fn reset_buffer(&mut self, buf: &'a [u8]) {
+ self.buf = buf;
+ self.last_read_field_id = 0;
+ self.read_field_id_stack.clear();
+ self.pending_read_bool_value = None;
+ }
+
+ pub fn as_slice(&self) -> &'a [u8] {
+ self.buf
+ }
+
+ fn read_vlq(&mut self) -> Result<u64> {
+ let mut in_progress = 0;
+ let mut shift = 0;
+ loop {
+ let byte = self.read_byte()?;
+ in_progress |= ((byte & 0x7F) as u64).wrapping_shl(shift);
+ shift += 7;
+ if byte & 0x80 == 0 {
+ return Ok(in_progress);
+ }
+ }
+ }
+
+ fn read_zig_zag(&mut self) -> Result<i64> {
+ let val = self.read_vlq()?;
+ Ok((val >> 1) as i64 ^ -((val & 1) as i64))
+ }
+
+ fn read_list_set_begin(&mut self) -> Result<(ElementType, i32)> {
+ let header = self.read_byte()?;
+ let element_type = ElementType::try_from(header & 0x0f)?;
+
+ let possible_element_count = (header & 0xF0) >> 4;
+ let element_count = if possible_element_count != 15 {
+ // high bits set high if count and type encoded separately
+ possible_element_count as i32
+ } else {
+ self.read_vlq()? as _
+ };
+
+ Ok((element_type, element_count))
+ }
+
+ pub(crate) fn read_struct_begin(&mut self) -> Result<()> {
+ self.read_field_id_stack.push(self.last_read_field_id);
+ self.last_read_field_id = 0;
+ Ok(())
+ }
+
+ pub(crate) fn read_struct_end(&mut self) -> Result<()> {
+ self.last_read_field_id = self
+ .read_field_id_stack
+ .pop()
+ .expect("should have previous field ids");
+ Ok(())
+ }
+
+ pub(crate) fn read_field_begin(&mut self) -> Result<FieldIdentifier> {
+ // we can read at least one byte, which is:
+ // - the type
+ // - the field delta and the type
+ let field_type = self.read_byte()?;
+ let field_delta = (field_type & 0xf0) >> 4;
+ let field_type = FieldType::try_from(field_type & 0xf)?;
+
+ match field_type {
+ FieldType::Stop => Ok(FieldIdentifier {
+ field_type: FieldType::Stop,
+ id: 0,
+ }),
+ _ => {
+ // special handling for bools
+ if field_type == FieldType::BooleanFalse {
+ self.pending_read_bool_value = Some(false);
+ } else if field_type == FieldType::BooleanTrue {
+ self.pending_read_bool_value = Some(true);
+ }
+ if field_delta != 0 {
+ self.last_read_field_id = self
+ .last_read_field_id
+ .checked_add(field_delta as i16)
+ .map_or_else(
+ || {
+ Err(general_err!(format!(
+ "cannot add {} to {}",
+ field_delta, self.last_read_field_id
+ )))
+ },
+ Ok,
+ )?;
+ } else {
+ self.last_read_field_id = self.read_i16()?;
+ };
+
+ Ok(FieldIdentifier {
+ field_type,
+ id: self.last_read_field_id,
+ })
+ }
+ }
+ }
+
+ pub(crate) fn read_bool(&mut self) -> Result<bool> {
+ match self.pending_read_bool_value.take() {
+ Some(b) => Ok(b),
+ None => {
+ let b = self.read_byte()?;
+ // Previous versions of the thrift specification said to use 0 and 1 inside collections,
+ // but that differed from existing implementations.
+ // The specification was updated in https://github.com/apache/thrift/commit/2c29c5665bc442e703480bb0ee60fe925ffe02e8.
+ // At least the go implementation seems to have followed the previously documented values.
+ match b {
+ 0x01 => Ok(true),
+ 0x00 | 0x02 => Ok(false),
+ unkn => Err(general_err!(format!("cannot convert {unkn} into bool"))),
+ }
+ }
+ }
+ }
+
+ pub(crate) fn read_bytes(&mut self) -> Result<&'b [u8]> {
+ let len = self.read_vlq()? as usize;
+ let ret = self.buf.get(..len).ok_or_else(eof_error)?;
+ self.buf = &self.buf[len..];
+ Ok(ret)
+ }
+
+ pub(crate) fn read_string(&mut self) -> Result<&'b str> {
+ let slice = self.read_bytes()?;
+ Ok(std::str::from_utf8(slice)?)
+ }
+
+ pub(crate) fn read_i8(&mut self) -> Result<i8> {
+ Ok(self.read_byte()? as _)
+ }
+
+ pub(crate) fn read_i16(&mut self) -> Result<i16> {
+ Ok(self.read_zig_zag()? as _)
+ }
+
+ pub(crate) fn read_i32(&mut self) -> Result<i32> {
+ Ok(self.read_zig_zag()? as _)
+ }
+
+ pub(crate) fn read_i64(&mut self) -> Result<i64> {
+ self.read_zig_zag()
+ }
+
+ pub(crate) fn read_double(&mut self) -> Result<f64> {
+ let slice = self.buf.get(..8).ok_or_else(eof_error)?;
+ self.buf = &self.buf[8..];
+ match slice.try_into() {
+ Ok(slice) => Ok(f64::from_le_bytes(slice)),
+ Err(_) => Err(general_err!("Unexpected error converting slice")),
+ }
+ }
+
+ pub(crate) fn read_list_begin(&mut self) -> Result<ListIdentifier> {
+ let (element_type, element_count) = self.read_list_set_begin()?;
+ Ok(ListIdentifier {
+ element_type,
+ size: element_count,
+ })
+ }
+
+ pub(crate) fn read_list_end(&mut self) -> Result<()> {
+ Ok(())
+ }
+
+ #[inline]
+ fn read_byte(&mut self) -> Result<u8> {
+ let ret = *self.buf.first().ok_or_else(eof_error)?;
+ self.buf = &self.buf[1..];
+ Ok(ret)
+ }
+
+ #[inline]
+ fn skip_bytes(&mut self, n: usize) -> Result<()> {
+ self.buf.get(..n).ok_or_else(eof_error)?;
+ self.buf = &self.buf[n..];
+ Ok(())
+ }
+
+ fn skip_vlq(&mut self) -> Result<()> {
+ loop {
+ let byte = self.read_byte()?;
+ if byte & 0x80 == 0 {
+ return Ok(());
+ }
+ }
+ }
+
+ /// Skip a field with type `field_type` recursively until the default
+ /// maximum skip depth is reached.
+ pub(crate) fn skip(&mut self, field_type: FieldType) -> Result<()> {
+ // TODO: magic number
+ self.skip_till_depth(field_type, 64)
+ }
+
+ /// Empty structs in unions consist of a single byte of 0 for the field stop record.
+ /// This skips that byte without pushing to the field id stack.
+ pub(crate) fn skip_empty_struct(&mut self) -> Result<()> {
+ let b = self.read_byte()?;
+ if b != 0 {
+ Err(general_err!("Empty struct has fields"))
+ } else {
+ Ok(())
+ }
+ }
+
+ /// Skip a field with type `field_type` recursively up to `depth` levels.
+ fn skip_till_depth(&mut self, field_type: FieldType, depth: i8) -> Result<()> {
+ if depth == 0 {
+ return Err(general_err!(format!("cannot parse past {:?}", field_type)));
+ }
+
+ match field_type {
+ FieldType::BooleanFalse | FieldType::BooleanTrue => self.read_bool().map(|_| ()),
+ FieldType::Byte => self.read_i8().map(|_| ()),
+ FieldType::I16 => self.skip_vlq().map(|_| ()),
+ FieldType::I32 => self.skip_vlq().map(|_| ()),
+ FieldType::I64 => self.skip_vlq().map(|_| ()),
+ FieldType::Double => self.skip_bytes(8).map(|_| ()),
+ FieldType::Binary => {
+ let len = self.read_vlq()? as usize;
+ self.skip_bytes(len)
+ }
+ FieldType::Struct => {
+ self.read_struct_begin()?;
+ loop {
+ let field_ident = self.read_field_begin()?;
+ if field_ident.field_type == FieldType::Stop {
+ break;
+ }
+ self.skip_till_depth(field_ident.field_type, depth - 1)?;
+ }
+ self.read_struct_end()
+ }
+ FieldType::List => {
+ let list_ident = self.read_list_begin()?;
+ for _ in 0..list_ident.size {
+ let element_type = FieldType::try_from(list_ident.element_type)?;
+ self.skip_till_depth(element_type, depth - 1)?;
+ }
+ self.read_list_end()
+ }
+ // no list or map types in parquet format
+ u => Err(general_err!(format!("cannot skip field type {:?}", &u))),
+ }
+ }
+}
+
+fn eof_error() -> ParquetError {
+ eof_err!("Unexpected EOF")
+}
diff --git a/parquet/src/schema/printer.rs b/parquet/src/schema/printer.rs
index 4190f9717d..fd28e13d2e 100644
--- a/parquet/src/schema/printer.rs
+++ b/parquet/src/schema/printer.rs
@@ -326,10 +326,15 @@ fn print_logical_and_converted(
LogicalType::List => "LIST".to_string(),
LogicalType::Map => "MAP".to_string(),
LogicalType::Float16 => "FLOAT16".to_string(),
- LogicalType::Variant => "VARIANT".to_string(),
- LogicalType::Geometry => "GEOMETRY".to_string(),
- LogicalType::Geography => "GEOGRAPHY".to_string(),
+ LogicalType::Variant {
+ specification_version,
+ } => format!("VARIANT({specification_version:?})"),
+ LogicalType::Geometry { crs } => format!("GEOMETRY({crs:?})"),
+ LogicalType::Geography { crs, algorithm } => {
+ format!("GEOGRAPHY({crs:?},{algorithm:?})")
+ }
LogicalType::Unknown => "UNKNOWN".to_string(),
+ LogicalType::_Unknown { field_id } => format!("_Unknown({field_id})"),
},
None => {
// Also print converted type if it is available
diff --git a/parquet/tests/arrow_reader/bad_data.rs b/parquet/tests/arrow_reader/bad_data.rs
index ba50e738f6..b340e97a3c 100644
--- a/parquet/tests/arrow_reader/bad_data.rs
+++ b/parquet/tests/arrow_reader/bad_data.rs
@@ -82,7 +82,7 @@ fn test_parquet_1481() {
let err = read_file("PARQUET-1481.parquet").unwrap_err();
assert_eq!(
err.to_string(),
- "Parquet error: unexpected parquet type: -7"
+ "Parquet error: Unexpected parquet Type: -7"
);
}