Skip to main content

parquet/file/metadata/thrift/
mod.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18//! This module is the bridge between a Parquet file's thrift encoded metadata
19//! and this crate's [Parquet metadata API]. It contains objects and functions used
20//! to serialize/deserialize metadata objects into/from the Thrift compact protocol
21//! format as defined by the [Parquet specification].
22//!
23//! [Parquet metadata API]: crate::file::metadata
24//! [Parquet specification]: https://github.com/apache/parquet-format/tree/master
25
26use std::io::Write;
27use std::sync::Arc;
28
29#[cfg(feature = "encryption")]
30pub(crate) mod encryption;
31
32#[cfg(feature = "encryption")]
33use crate::file::{
34    column_crypto_metadata::ColumnCryptoMetaData, metadata::thrift::encryption::EncryptionAlgorithm,
35};
36use crate::{
37    basic::{
38        ColumnOrder, CompressionCodec, ConvertedType, Encoding, EncodingMask, LogicalType,
39        PageType, Repetition, Type,
40    },
41    data_type::{ByteArray, FixedLenByteArray, Int96},
42    errors::{ParquetError, Result},
43    file::{
44        metadata::{
45            ColumnChunkMetaData, ColumnChunkMetaDataBuilder, KeyValue, LevelHistogram,
46            PageEncodingStats, ParquetMetaData, ParquetMetaDataOptions, ParquetPageEncodingStats,
47            RowGroupMetaData, RowGroupMetaDataBuilder, SortingColumn,
48        },
49        statistics::ValueStatistics,
50    },
51    parquet_thrift::{
52        ElementType, FieldType, ReadThrift, ThriftCompactInputProtocol,
53        ThriftCompactOutputProtocol, ThriftSliceInputProtocol, WriteThrift, WriteThriftField,
54        check_list_size, read_thrift_vec, validate_list_type,
55    },
56    schema::types::{
57        ColumnDescriptor, SchemaDescriptor, TypePtr, num_nodes, parquet_schema_from_array,
58    },
59    thrift_struct,
60    util::bit_util::FromBytes,
61    write_thrift_field,
62};
63
64// this needs to be visible to the schema conversion code
65thrift_struct!(
66pub(crate) struct SchemaElement<'a> {
67  /// Data type for this field. Not set if the current element is a non-leaf node
68  1: optional Type r#type;
69  /// If type is FIXED_LEN_BYTE_ARRAY, this is the byte length of the values.
70  /// Otherwise, if specified, this is the maximum bit length to store any of the values.
71  /// (e.g. a low cardinality INT col could have this set to 3).  Note that this is
72  /// in the schema, and therefore fixed for the entire file.
73  2: optional i32 type_length;
74  /// Repetition of the field. The root of the schema does not have a repetition_type.
75  /// All other nodes must have one.
76  3: optional Repetition repetition_type;
77  /// Name of the field in the schema
78  4: required string<'a> name;
79  /// Nested fields. Since thrift does not support nested fields,
80  /// the nesting is flattened to a single list by a depth-first traversal.
81  /// The children count is used to construct the nested relationship.
82  /// This field is not set when the element is a primitive type.
83  5: optional i32 num_children;
84  /// DEPRECATED: When the schema is the result of a conversion from another model.
85  /// Used to record the original type to help with cross conversion.
86  ///
87  /// This is superseded by logical_type.
88  6: optional ConvertedType converted_type;
89  /// DEPRECATED: Used when this column contains decimal data.
90  /// See the DECIMAL converted type for more details.
91  ///
92  /// This is superseded by using the DecimalType annotation in logical_type.
93  7: optional i32 scale
94  8: optional i32 precision
95  /// When the original schema supports field ids, this will save the
96  /// original field id in the parquet schema
97  9: optional i32 field_id;
98  /// The logical type of this SchemaElement
99  ///
100  /// LogicalType replaces ConvertedType, but ConvertedType is still required
101  /// for some logical types to ensure forward-compatibility in format v1.
102  10: optional LogicalType logical_type
103}
104);
105
106thrift_struct!(
107struct Statistics<'a> {
108   1: optional binary<'a> max;
109   2: optional binary<'a> min;
110   3: optional i64 null_count;
111   4: optional i64 distinct_count;
112   5: optional binary<'a> max_value;
113   6: optional binary<'a> min_value;
114   7: optional bool is_max_value_exact;
115   8: optional bool is_min_value_exact;
116   9: optional i64 nan_count;
117}
118);
119
120thrift_struct!(
121struct BoundingBox {
122  1: required double xmin;
123  2: required double xmax;
124  3: required double ymin;
125  4: required double ymax;
126  5: optional double zmin;
127  6: optional double zmax;
128  7: optional double mmin;
129  8: optional double mmax;
130}
131);
132
133thrift_struct!(
134struct GeospatialStatistics {
135  1: optional BoundingBox bbox;
136  2: optional list<i32> geospatial_types;
137}
138);
139
140thrift_struct!(
141struct SizeStatistics {
142   1: optional i64 unencoded_byte_array_data_bytes;
143   2: optional list<i64> repetition_level_histogram;
144   3: optional list<i64> definition_level_histogram;
145}
146);
147
148fn convert_geo_stats(
149    st: GeospatialStatistics,
150) -> crate::geospatial::statistics::GeospatialStatistics {
151    let bbox = st.bbox.map(convert_bounding_box);
152    let geospatial_types: Option<Vec<i32>> = st.geospatial_types.filter(|v| !v.is_empty());
153    crate::geospatial::statistics::GeospatialStatistics::new(bbox, geospatial_types)
154}
155
156fn convert_bounding_box(bb: BoundingBox) -> crate::geospatial::bounding_box::BoundingBox {
157    let mut newbb = crate::geospatial::bounding_box::BoundingBox::new(
158        bb.xmin.into(),
159        bb.xmax.into(),
160        bb.ymin.into(),
161        bb.ymax.into(),
162    );
163
164    newbb = match (bb.zmin, bb.zmax) {
165        (Some(zmin), Some(zmax)) => newbb.with_zrange(zmin.into(), zmax.into()),
166        // If either None or mismatch, leave it as None and don't error
167        _ => newbb,
168    };
169
170    newbb = match (bb.mmin, bb.mmax) {
171        (Some(mmin), Some(mmax)) => newbb.with_mrange(mmin.into(), mmax.into()),
172        // If either None or mismatch, leave it as None and don't error
173        _ => newbb,
174    };
175
176    newbb
177}
178
179/// Create a [`crate::file::statistics::Statistics`] from a thrift [`Statistics`] object.
180fn convert_stats(
181    column_descr: &Arc<ColumnDescriptor>,
182    thrift_stats: Option<Statistics>,
183) -> Result<Option<crate::file::statistics::Statistics>> {
184    use crate::file::statistics::Statistics as FStatistics;
185    Ok(match thrift_stats {
186        Some(stats) => {
187            // Generic null count.
188            let null_count = stats
189                .null_count
190                .map(|null_count| {
191                    if null_count < 0 {
192                        return Err(general_err!(
193                            "Statistics null count is negative {}",
194                            null_count
195                        ));
196                    }
197                    Ok(null_count as u64)
198                })
199                .transpose()?;
200            // Generic distinct count (count of distinct values occurring)
201            let distinct_count = stats.distinct_count.map(|value| value as u64);
202            // Generic nan count for floating point types
203            let nan_count = stats
204                .nan_count
205                .map(|nan_count| {
206                    if nan_count < 0 {
207                        return Err(general_err!(
208                            "Statistics NaN count is negative {}",
209                            nan_count
210                        ));
211                    }
212                    Ok(nan_count as u64)
213                })
214                .transpose()?;
215            // Whether or not statistics use deprecated min/max fields.
216            let old_format = stats.min_value.is_none() && stats.max_value.is_none();
217            // Generic min value as bytes.
218            let min = if old_format {
219                stats.min
220            } else {
221                stats.min_value
222            };
223            // Generic max value as bytes.
224            let max = if old_format {
225                stats.max
226            } else {
227                stats.max_value
228            };
229
230            fn check_len(min: Option<&[u8]>, max: Option<&[u8]>, len: usize) -> Result<()> {
231                if let Some(min) = min
232                    && min.len() < len
233                {
234                    return Err(general_err!("Insufficient bytes to parse min statistic",));
235                }
236                if let Some(max) = max
237                    && max.len() < len
238                {
239                    return Err(general_err!("Insufficient bytes to parse max statistic",));
240                }
241                Ok(())
242            }
243
244            let physical_type = column_descr.physical_type();
245            match physical_type {
246                Type::BOOLEAN => check_len(min, max, 1),
247                Type::INT32 | Type::FLOAT => check_len(min, max, 4),
248                Type::INT64 | Type::DOUBLE => check_len(min, max, 8),
249                Type::INT96 => check_len(min, max, 12),
250                _ => Ok(()),
251            }?;
252
253            // Values are encoded using PLAIN encoding definition, except that
254            // variable-length byte arrays do not include a length prefix.
255            //
256            // Instead of using actual decoder, we manually convert values.
257            let res = match physical_type {
258                Type::BOOLEAN => FStatistics::boolean(
259                    min.map(|data| data[0] != 0),
260                    max.map(|data| data[0] != 0),
261                    distinct_count,
262                    null_count,
263                    old_format,
264                ),
265                Type::INT32 => FStatistics::int32(
266                    min.map(|data| i32::from_le_bytes(data[..4].try_into().unwrap())),
267                    max.map(|data| i32::from_le_bytes(data[..4].try_into().unwrap())),
268                    distinct_count,
269                    null_count,
270                    old_format,
271                ),
272                Type::INT64 => FStatistics::int64(
273                    min.map(|data| i64::from_le_bytes(data[..8].try_into().unwrap())),
274                    max.map(|data| i64::from_le_bytes(data[..8].try_into().unwrap())),
275                    distinct_count,
276                    null_count,
277                    old_format,
278                ),
279                Type::INT96 => {
280                    // INT96 statistics may not be correct, because comparison is signed
281                    let min = if let Some(data) = min {
282                        if data.len() != 12 {
283                            return Err(general_err!("Incorrect Int96 min statistics"));
284                        }
285                        Some(Int96::try_from_le_slice(data)?)
286                    } else {
287                        None
288                    };
289                    let max = if let Some(data) = max {
290                        if data.len() != 12 {
291                            return Err(general_err!("Incorrect Int96 max statistics"));
292                        }
293                        Some(Int96::try_from_le_slice(data)?)
294                    } else {
295                        None
296                    };
297                    FStatistics::int96(min, max, distinct_count, null_count, old_format)
298                }
299                Type::FLOAT => FStatistics::Float(
300                    ValueStatistics::new(
301                        min.map(|data| f32::from_le_bytes(data[..4].try_into().unwrap())),
302                        max.map(|data| f32::from_le_bytes(data[..4].try_into().unwrap())),
303                        distinct_count,
304                        null_count,
305                        old_format,
306                    )
307                    .with_nan_count(nan_count),
308                ),
309                Type::DOUBLE => FStatistics::Double(
310                    ValueStatistics::new(
311                        min.map(|data| f64::from_le_bytes(data[..8].try_into().unwrap())),
312                        max.map(|data| f64::from_le_bytes(data[..8].try_into().unwrap())),
313                        distinct_count,
314                        null_count,
315                        old_format,
316                    )
317                    .with_nan_count(nan_count),
318                ),
319                Type::BYTE_ARRAY => FStatistics::ByteArray(
320                    ValueStatistics::new(
321                        min.map(ByteArray::from),
322                        max.map(ByteArray::from),
323                        distinct_count,
324                        null_count,
325                        old_format,
326                    )
327                    .with_max_is_exact(stats.is_max_value_exact.unwrap_or(false))
328                    .with_min_is_exact(stats.is_min_value_exact.unwrap_or(false)),
329                ),
330                Type::FIXED_LEN_BYTE_ARRAY => FStatistics::FixedLenByteArray(
331                    ValueStatistics::new(
332                        min.map(ByteArray::from).map(FixedLenByteArray::from),
333                        max.map(ByteArray::from).map(FixedLenByteArray::from),
334                        distinct_count,
335                        null_count,
336                        old_format,
337                    )
338                    .with_nan_count(nan_count)
339                    .with_max_is_exact(stats.is_max_value_exact.unwrap_or(false))
340                    .with_min_is_exact(stats.is_min_value_exact.unwrap_or(false)),
341                ),
342            };
343
344            Some(res)
345        }
346        None => None,
347    })
348}
349
350// bit positions for required fields in the Thrift ColumnMetaData struct
351const COL_META_TYPE: u16 = 1 << 1;
352const COL_META_ENCODINGS: u16 = 1 << 2;
353const COL_META_CODEC: u16 = 1 << 4;
354const COL_META_NUM_VALUES: u16 = 1 << 5;
355const COL_META_TOTAL_UNCOMP_SZ: u16 = 1 << 6;
356const COL_META_TOTAL_COMP_SZ: u16 = 1 << 7;
357const COL_META_DATA_PAGE_OFFSET: u16 = 1 << 9;
358
359// a mask where all required fields' bits are set
360const COL_META_ALL_REQUIRED: u16 = COL_META_TYPE
361    | COL_META_ENCODINGS
362    | COL_META_CODEC
363    | COL_META_NUM_VALUES
364    | COL_META_TOTAL_UNCOMP_SZ
365    | COL_META_TOTAL_COMP_SZ
366    | COL_META_DATA_PAGE_OFFSET;
367
368// check mask to see if all required fields are set. return an appropriate error if
369// any are missing.
370fn validate_column_metadata(mask: u16) -> Result<()> {
371    if mask != COL_META_ALL_REQUIRED {
372        if mask & COL_META_ENCODINGS == 0 {
373            return Err(general_err!("Required field encodings is missing"));
374        }
375
376        if mask & COL_META_CODEC == 0 {
377            return Err(general_err!("Required field codec is missing"));
378        }
379        if mask & COL_META_NUM_VALUES == 0 {
380            return Err(general_err!("Required field num_values is missing"));
381        }
382        if mask & COL_META_TOTAL_UNCOMP_SZ == 0 {
383            return Err(general_err!(
384                "Required field total_uncompressed_size is missing"
385            ));
386        }
387        if mask & COL_META_TOTAL_COMP_SZ == 0 {
388            return Err(general_err!(
389                "Required field total_compressed_size is missing"
390            ));
391        }
392        if mask & COL_META_DATA_PAGE_OFFSET == 0 {
393            return Err(general_err!("Required field data_page_offset is missing"));
394        }
395    }
396
397    Ok(())
398}
399
400fn read_encoding_stats_as_mask(prot: &mut ThriftSliceInputProtocol<'_>) -> Result<EncodingMask> {
401    // read the vector of stats, setting mask bits for data pages
402    let mut mask = 0i32;
403    let list_ident = prot.read_list_begin()?;
404    // check for PageEncodingStats struct
405    validate_list_type(ElementType::Struct, &list_ident)?;
406
407    for _ in 0..list_ident.size {
408        let pes = PageEncodingStats::read_thrift(prot)?;
409        match pes.page_type {
410            PageType::DATA_PAGE | PageType::DATA_PAGE_V2 => mask |= 1 << pes.encoding as i32,
411            _ => {}
412        }
413    }
414    EncodingMask::try_new(mask)
415}
416
417// Decode `ColumnMetaData`. Returns a mask of all required fields that were observed.
418// This mask can be passed to `validate_column_metadata`.
419#[expect(clippy::useless_let_if_seq)] // the `let mut … if let …` below is more readable than the suggestion
420fn read_column_metadata(
421    prot: &mut ThriftSliceInputProtocol<'_>,
422    column: &mut ColumnChunkMetaData,
423    col_index: usize,
424    options: Option<&ParquetMetaDataOptions>,
425) -> Result<u16> {
426    // mask for seen required fields in ColumnMetaData
427    let mut seen_mask = 0u16;
428
429    let mut skip_pes = false;
430    let mut pes_mask = true;
431    let mut skip_col_stats = false;
432    let mut skip_size_stats = false;
433
434    if let Some(opts) = options {
435        skip_pes = opts.skip_encoding_stats(col_index);
436        pes_mask = opts.encoding_stats_as_mask();
437        skip_col_stats = opts.skip_column_stats(col_index);
438        skip_size_stats = opts.skip_size_stats(col_index);
439    }
440
441    // struct ColumnMetaData {
442    //   1: required Type type
443    //   2: required list<Encoding> encodings
444    //   3: required list<string> path_in_schema
445    //   4: required CompressionCodec codec
446    //   5: required i64 num_values
447    //   6: required i64 total_uncompressed_size
448    //   7: required i64 total_compressed_size
449    //   8: optional list<KeyValue> key_value_metadata
450    //   9: required i64 data_page_offset
451    //   10: optional i64 index_page_offset
452    //   11: optional i64 dictionary_page_offset
453    //   12: optional Statistics statistics;
454    //   13: optional list<PageEncodingStats> encoding_stats;
455    //   14: optional i64 bloom_filter_offset;
456    //   15: optional i32 bloom_filter_length;
457    //   16: optional SizeStatistics size_statistics;
458    //   17: optional GeospatialStatistics geospatial_statistics;
459    // }
460    let column_descr = &column.column_descr;
461
462    let mut last_field_id = 0i16;
463    loop {
464        let field_ident = prot.read_field_begin(last_field_id)?;
465        if field_ident.field_type == FieldType::Stop {
466            break;
467        }
468        match field_ident.id {
469            // 1: type is never used, we can use the column descriptor
470            1 => {
471                // read for error handling
472                Type::read_thrift(&mut *prot)?;
473                seen_mask |= COL_META_TYPE;
474            }
475            2 => {
476                column.encodings = EncodingMask::read_thrift(&mut *prot)?;
477                seen_mask |= COL_META_ENCODINGS;
478            }
479            // 3: path_in_schema is redundant
480            4 => {
481                column.compression = CompressionCodec::read_thrift(&mut *prot)?;
482                seen_mask |= COL_META_CODEC;
483            }
484            5 => {
485                column.num_values = i64::read_thrift(&mut *prot)?;
486                seen_mask |= COL_META_NUM_VALUES;
487            }
488            6 => {
489                column.total_uncompressed_size = i64::read_thrift(&mut *prot)?;
490                seen_mask |= COL_META_TOTAL_UNCOMP_SZ;
491            }
492            7 => {
493                column.total_compressed_size = i64::read_thrift(&mut *prot)?;
494                seen_mask |= COL_META_TOTAL_COMP_SZ;
495            }
496            // 8: we don't expose this key value
497            9 => {
498                column.data_page_offset = i64::read_thrift(&mut *prot)?;
499                seen_mask |= COL_META_DATA_PAGE_OFFSET;
500            }
501            10 => {
502                column.index_page_offset = Some(i64::read_thrift(&mut *prot)?);
503            }
504            11 => {
505                column.dictionary_page_offset = Some(i64::read_thrift(&mut *prot)?);
506            }
507            12 if !skip_col_stats => {
508                column.statistics =
509                    convert_stats(column_descr, Some(Statistics::read_thrift(&mut *prot)?))?;
510            }
511            13 if !skip_pes => {
512                if pes_mask {
513                    let val = read_encoding_stats_as_mask(&mut *prot)?;
514                    column.encoding_stats = Some(ParquetPageEncodingStats::Mask(val));
515                } else {
516                    let val =
517                        read_thrift_vec::<PageEncodingStats, ThriftSliceInputProtocol>(&mut *prot)?;
518                    column.encoding_stats = Some(ParquetPageEncodingStats::Full(val));
519                }
520            }
521            14 => {
522                column.bloom_filter_offset = Some(i64::read_thrift(&mut *prot)?);
523            }
524            15 => {
525                column.bloom_filter_length = Some(i32::read_thrift(&mut *prot)?);
526            }
527            16 if !skip_size_stats => {
528                let val = SizeStatistics::read_thrift(&mut *prot)?;
529                column.unencoded_byte_array_data_bytes = val.unencoded_byte_array_data_bytes;
530                column.repetition_level_histogram =
531                    val.repetition_level_histogram.map(LevelHistogram::from);
532                column.definition_level_histogram =
533                    val.definition_level_histogram.map(LevelHistogram::from);
534            }
535            17 => {
536                let val = GeospatialStatistics::read_thrift(&mut *prot)?;
537                column.geo_statistics = Some(Box::new(convert_geo_stats(val)));
538            }
539            _ => {
540                prot.skip(field_ident.field_type)?;
541            }
542        }
543        last_field_id = field_ident.id;
544    }
545
546    Ok(seen_mask)
547}
548
549// using ThriftSliceInputProtocol rather than ThriftCompactInputProtocol trait because
550// these are all internal and operate on slices.
551fn read_column_chunk(
552    prot: &mut ThriftSliceInputProtocol<'_>,
553    column_descr: &Arc<ColumnDescriptor>,
554    col_index: usize,
555    options: Option<&ParquetMetaDataOptions>,
556) -> Result<ColumnChunkMetaData> {
557    // create a default initialized ColumnMetaData
558    let mut col = ColumnChunkMetaDataBuilder::new(column_descr.clone()).build()?;
559
560    // seen flag for file_offset
561    let mut has_file_offset = false;
562
563    // mask of seen flags for ColumnMetaData
564    let mut col_meta_mask = 0u16;
565
566    // struct ColumnChunk {
567    //   1: optional string file_path
568    //   2: required i64 file_offset = 0
569    //   3: optional ColumnMetaData meta_data
570    //   4: optional i64 offset_index_offset
571    //   5: optional i32 offset_index_length
572    //   6: optional i64 column_index_offset
573    //   7: optional i32 column_index_length
574    //   8: optional ColumnCryptoMetaData crypto_metadata
575    //   9: optional binary encrypted_column_metadata
576    // }
577    let mut last_field_id = 0i16;
578    loop {
579        let field_ident = prot.read_field_begin(last_field_id)?;
580        if field_ident.field_type == FieldType::Stop {
581            break;
582        }
583        match field_ident.id {
584            1 => {
585                col.file_path = Some(String::read_thrift(&mut *prot)?);
586            }
587            2 => {
588                col.file_offset = i64::read_thrift(&mut *prot)?;
589                has_file_offset = true;
590            }
591            3 => {
592                col_meta_mask = read_column_metadata(&mut *prot, &mut col, col_index, options)?;
593            }
594            4 => {
595                col.offset_index_offset = Some(i64::read_thrift(&mut *prot)?);
596            }
597            5 => {
598                col.offset_index_length = Some(i32::read_thrift(&mut *prot)?);
599            }
600            6 => {
601                col.column_index_offset = Some(i64::read_thrift(&mut *prot)?);
602            }
603            7 => {
604                col.column_index_length = Some(i32::read_thrift(&mut *prot)?);
605            }
606            #[cfg(feature = "encryption")]
607            8 => {
608                let val = ColumnCryptoMetaData::read_thrift(&mut *prot)?;
609                col.column_crypto_metadata = Some(Box::new(val));
610            }
611            #[cfg(feature = "encryption")]
612            9 => {
613                col.encrypted_column_metadata = Some(<&[u8]>::read_thrift(&mut *prot)?.to_vec());
614            }
615            _ => {
616                prot.skip(field_ident.field_type)?;
617            }
618        }
619        last_field_id = field_ident.id;
620    }
621
622    // the only required field from ColumnChunk
623    if !has_file_offset {
624        return Err(general_err!("Required field file_offset is missing"));
625    }
626
627    // if encrypted just return. we'll decrypt after finishing the footer and populate the rest.
628    #[cfg(feature = "encryption")]
629    if col.encrypted_column_metadata.is_some() {
630        return Ok(col);
631    }
632
633    // not encrypted, so make sure all required fields were read
634    validate_column_metadata(col_meta_mask)?;
635
636    Ok(col)
637}
638
639fn read_row_group(
640    prot: &mut ThriftSliceInputProtocol,
641    schema_descr: &Arc<SchemaDescriptor>,
642    options: Option<&ParquetMetaDataOptions>,
643) -> Result<RowGroupMetaData> {
644    // create default initialized RowGroupMetaData
645    let mut row_group = RowGroupMetaDataBuilder::new(schema_descr.clone()).build_unchecked();
646
647    // mask values for required fields
648    const RG_COLUMNS: u8 = 1 << 1;
649    const RG_TOT_BYTE_SIZE: u8 = 1 << 2;
650    const RG_NUM_ROWS: u8 = 1 << 3;
651    const RG_ALL_REQUIRED: u8 = RG_COLUMNS | RG_TOT_BYTE_SIZE | RG_NUM_ROWS;
652
653    let mut mask = 0u8;
654
655    // struct RowGroup {
656    //   1: required list<ColumnChunk> columns
657    //   2: required i64 total_byte_size
658    //   3: required i64 num_rows
659    //   4: optional list<SortingColumn> sorting_columns
660    //   5: optional i64 file_offset
661    //   6: optional i64 total_compressed_size
662    //   7: optional i16 ordinal
663    // }
664    let mut last_field_id = 0i16;
665    loop {
666        let field_ident = prot.read_field_begin(last_field_id)?;
667        if field_ident.field_type == FieldType::Stop {
668            break;
669        }
670        match field_ident.id {
671            1 => {
672                let list_ident = prot.read_list_begin()?;
673                // check for list of struct
674                validate_list_type(ElementType::Struct, &list_ident)?;
675                if schema_descr.num_columns() != list_ident.size as usize {
676                    return Err(general_err!(
677                        "Column count mismatch. Schema has {} columns while Row Group has {}",
678                        schema_descr.num_columns(),
679                        list_ident.size
680                    ));
681                }
682                for i in 0..list_ident.size as usize {
683                    let col = read_column_chunk(prot, &schema_descr.columns()[i], i, options)?;
684                    row_group.columns.push(col);
685                }
686                mask |= RG_COLUMNS;
687            }
688            2 => {
689                row_group.total_byte_size = i64::read_thrift(&mut *prot)?;
690                mask |= RG_TOT_BYTE_SIZE;
691            }
692            3 => {
693                row_group.num_rows = i64::read_thrift(&mut *prot)?;
694                mask |= RG_NUM_ROWS;
695            }
696            4 => {
697                let val = read_thrift_vec::<SortingColumn, ThriftSliceInputProtocol>(&mut *prot)?;
698                row_group.sorting_columns = Some(val);
699            }
700            5 => {
701                row_group.file_offset = Some(i64::read_thrift(&mut *prot)?);
702            }
703            // 6: we don't expose total_compressed_size
704            7 => {
705                row_group.ordinal = Some(i16::read_thrift(&mut *prot)? as i32);
706            }
707            _ => {
708                prot.skip(field_ident.field_type)?;
709            }
710        }
711        last_field_id = field_ident.id;
712    }
713
714    if mask != RG_ALL_REQUIRED {
715        if mask & RG_COLUMNS == 0 {
716            return Err(general_err!("Required field columns is missing"));
717        }
718        if mask & RG_TOT_BYTE_SIZE == 0 {
719            return Err(general_err!("Required field total_byte_size is missing"));
720        }
721        if mask & RG_NUM_ROWS == 0 {
722            return Err(general_err!("Required field num_rows is missing"));
723        }
724    }
725
726    Ok(row_group)
727}
728
729/// Create a [`SchemaDescriptor`] from thrift input. The input buffer must contain a complete
730/// Parquet footer.
731pub(crate) fn parquet_schema_from_bytes(buf: &[u8]) -> Result<SchemaDescriptor> {
732    let mut prot = ThriftSliceInputProtocol::new(buf);
733
734    let mut last_field_id = 0i16;
735    loop {
736        let field_ident = prot.read_field_begin(last_field_id)?;
737        if field_ident.field_type == FieldType::Stop {
738            break;
739        }
740        match field_ident.id {
741            2 => {
742                // read schema and convert to SchemaDescriptor for use when reading row groups
743                let val = read_thrift_vec::<SchemaElement, ThriftSliceInputProtocol>(&mut prot)?;
744                let val = parquet_schema_from_array(val)?;
745                return Ok(SchemaDescriptor::new(val));
746            }
747            _ => prot.skip(field_ident.field_type)?,
748        }
749        last_field_id = field_ident.id;
750    }
751    Err(general_err!("Input does not contain a schema"))
752}
753
754/// Create [`ParquetMetaData`] from thrift input. Note that this only decodes the file metadata in
755/// the Parquet footer. Page indexes will need to be added later.
756pub(crate) fn parquet_metadata_from_bytes(
757    buf: &[u8],
758    options: Option<&ParquetMetaDataOptions>,
759) -> Result<ParquetMetaData> {
760    let mut prot = ThriftSliceInputProtocol::new(buf);
761
762    // begin reading the file metadata
763    let mut version: Option<i32> = None;
764    let mut num_rows: Option<i64> = None;
765    let mut row_groups: Option<Vec<RowGroupMetaData>> = None;
766    let mut key_value_metadata: Option<Vec<KeyValue>> = None;
767    let mut created_by: Option<&str> = None;
768    let mut column_orders: Option<Vec<ColumnOrder>> = None;
769    #[cfg(feature = "encryption")]
770    let mut encryption_algorithm: Option<EncryptionAlgorithm> = None;
771    #[cfg(feature = "encryption")]
772    let mut footer_signing_key_metadata: Option<&[u8]> = None;
773
774    // this will need to be set before parsing row groups.
775    // see if we already have a schema.
776    let mut schema_descr: Option<Arc<SchemaDescriptor>> =
777        options.and_then(|options| options.schema().cloned());
778
779    // struct FileMetaData {
780    //   1: required i32 version
781    //   2: required list<SchemaElement> schema;
782    //   3: required i64 num_rows
783    //   4: required list<RowGroup> row_groups
784    //   5: optional list<KeyValue> key_value_metadata
785    //   6: optional string created_by
786    //   7: optional list<ColumnOrder> column_orders;
787    //   8: optional EncryptionAlgorithm encryption_algorithm
788    //   9: optional binary footer_signing_key_metadata
789    // }
790    let mut last_field_id = 0i16;
791    loop {
792        let field_ident = prot.read_field_begin(last_field_id)?;
793        if field_ident.field_type == FieldType::Stop {
794            break;
795        }
796        match field_ident.id {
797            1 => {
798                version = Some(i32::read_thrift(&mut prot)?);
799            }
800            2 => {
801                // If schema was passed in, skip parsing it
802                if schema_descr.is_some() {
803                    prot.skip(field_ident.field_type)?;
804                } else {
805                    // read schema and convert to SchemaDescriptor for use when reading row groups
806                    let val =
807                        read_thrift_vec::<SchemaElement, ThriftSliceInputProtocol>(&mut prot)?;
808                    let val = parquet_schema_from_array(val)?;
809                    schema_descr = Some(Arc::new(SchemaDescriptor::new(val)));
810                }
811            }
812            3 => {
813                num_rows = Some(i64::read_thrift(&mut prot)?);
814            }
815            4 => {
816                if schema_descr.is_none() {
817                    return Err(general_err!("Required field schema is missing"));
818                }
819                let schema_descr = schema_descr.as_ref().unwrap();
820                let list_ident = prot.read_list_begin()?;
821                // check for list of struct
822                validate_list_type(ElementType::Struct, &list_ident)?;
823                let size = list_ident.size as usize;
824                // Validate list size.
825                check_list_size(prot.remaining_bytes(), size)?;
826                let mut rg_vec = Vec::with_capacity(size);
827
828                for _ in 0..list_ident.size {
829                    rg_vec.push(read_row_group(&mut prot, schema_descr, options)?);
830                }
831                ensure_row_group_ordinals(&mut rg_vec)?;
832                row_groups = Some(rg_vec);
833            }
834            5 => {
835                let val = read_thrift_vec::<KeyValue, ThriftSliceInputProtocol>(&mut prot)?;
836                key_value_metadata = Some(val);
837            }
838            6 => {
839                created_by = Some(<&str>::read_thrift(&mut prot)?);
840            }
841            7 => {
842                let val = read_thrift_vec::<ColumnOrder, ThriftSliceInputProtocol>(&mut prot)?;
843                column_orders = Some(val);
844            }
845            #[cfg(feature = "encryption")]
846            8 => {
847                let val = EncryptionAlgorithm::read_thrift(&mut prot)?;
848                encryption_algorithm = Some(val);
849            }
850            #[cfg(feature = "encryption")]
851            9 => {
852                footer_signing_key_metadata = Some(<&[u8]>::read_thrift(&mut prot)?);
853            }
854            _ => {
855                prot.skip(field_ident.field_type)?;
856            }
857        }
858        last_field_id = field_ident.id;
859    }
860    let Some(version) = version else {
861        return Err(general_err!("Required field version is missing"));
862    };
863    let Some(num_rows) = num_rows else {
864        return Err(general_err!("Required field num_rows is missing"));
865    };
866    let Some(row_groups) = row_groups else {
867        return Err(general_err!("Required field row_groups is missing"));
868    };
869
870    let created_by = created_by.map(|c| c.to_owned());
871
872    // we've tested for `None` by now so this is safe
873    let schema_descr = schema_descr.unwrap();
874
875    // need to map read column orders to actual values based on the schema
876    if column_orders
877        .as_ref()
878        .is_some_and(|cos| cos.len() != schema_descr.num_columns())
879    {
880        return Err(general_err!("Column order length mismatch"));
881    }
882    // replace default type defined column orders with ones having the correct sort order
883    let column_orders = column_orders.map(|mut cos| {
884        for (i, column) in schema_descr.columns().iter().enumerate() {
885            if let ColumnOrder::TYPE_DEFINED_ORDER(_) = cos[i] {
886                // use `get_sort_order_for_type` so we don't replace a type defined sort order
887                // with a more recent ordering. we need to preserve what was actually in the
888                // footer.
889                let sort_order = ColumnOrder::get_sort_order_for_type(
890                    column.logical_type_ref(),
891                    column.converted_type(),
892                    column.physical_type(),
893                    true,
894                );
895                cos[i] = ColumnOrder::TYPE_DEFINED_ORDER(sort_order);
896            }
897        }
898        cos
899    });
900
901    #[cfg(not(feature = "encryption"))]
902    let fmd = crate::file::metadata::FileMetaData::new(
903        version,
904        num_rows,
905        created_by,
906        key_value_metadata,
907        schema_descr,
908        column_orders,
909    );
910    #[cfg(feature = "encryption")]
911    let fmd = crate::file::metadata::FileMetaData::new(
912        version,
913        num_rows,
914        created_by,
915        key_value_metadata,
916        schema_descr,
917        column_orders,
918    )
919    .with_encryption_algorithm(encryption_algorithm)
920    .with_footer_signing_key_metadata(footer_signing_key_metadata.map(|v| v.to_vec()));
921
922    Ok(ParquetMetaData::new(fmd, row_groups))
923}
924
925/// Ensure [`RowGroupMetaData::ordinal`] is usable after decode without
926/// rejecting spec-valid files (`RowGroup.ordinal` is optional in the
927/// parquet-format Thrift definition, with no uniformity requirement):
928///
929/// - **All row groups carry ordinals** → honor them as written.
930/// - **No row group carries an ordinal** → assign each row group its
931///   position in the file. This happens unconditionally (not only when a
932///   consumer needs it) so downstream users of the ordinal — the row
933///   number virtual column, encryption chunk-key lookup — behave the same
934///   whether the metadata was decoded fresh or reused from a prior read.
935/// - **Mixed** → leave the metadata untouched. Positional backfill could
936///   disagree with the ordinals that are present, and a partial backfill
937///   would make row-number results depend on which row groups a query
938///   happens to select. Consumers that require complete ordinals fail
939///   deterministically instead (see `RowNumberReader::try_new`); plain
940///   reads that never touch ordinals succeed.
941fn ensure_row_group_ordinals(row_groups: &mut [RowGroupMetaData]) -> Result<()> {
942    // All set (honor them as written) or mixed (leave as-is): either way there
943    // is nothing to backfill. Only when *no* row group carries an ordinal do we
944    // assign positions below.
945    if row_groups.iter().any(|rg| rg.ordinal.is_some()) {
946        return Ok(());
947    }
948    for (idx, rg) in row_groups.iter_mut().enumerate() {
949        let ordinal: i32 = idx
950            .try_into()
951            .map_err(|_| general_err!("Row group ordinal {} exceeds i32 max value", idx))?;
952        rg.ordinal = Some(ordinal);
953    }
954    Ok(())
955}
956
957thrift_struct!(
958    pub(crate) struct IndexPageHeader {}
959);
960
961thrift_struct!(
962pub(crate) struct DictionaryPageHeader {
963  /// Number of values in the dictionary
964  1: required i32 num_values;
965
966  /// Encoding using this dictionary page
967  2: required Encoding encoding
968
969  /// If true, the entries in the dictionary are sorted in ascending order
970  3: optional bool is_sorted;
971}
972);
973
974thrift_struct!(
975/// Statistics for the page header.
976///
977/// This is a duplicate of the [`Statistics`] struct above. Because the page reader uses
978/// the [`Read`] API, we cannot read the min/max values as slices. This should not be
979/// a huge problem since this crate no longer reads the page header statistics by default.
980///
981/// [`Read`]: crate::parquet_thrift::ThriftReadInputProtocol
982pub(crate) struct PageStatistics {
983   1: optional binary max;
984   2: optional binary min;
985   3: optional i64 null_count;
986   4: optional i64 distinct_count;
987   5: optional binary max_value;
988   6: optional binary min_value;
989   7: optional bool is_max_value_exact;
990   8: optional bool is_min_value_exact;
991   9: optional i64 nan_count;
992}
993);
994
995thrift_struct!(
996pub(crate) struct DataPageHeader {
997  1: required i32 num_values
998  2: required Encoding encoding
999  3: required Encoding definition_level_encoding;
1000  4: required Encoding repetition_level_encoding;
1001  5: optional PageStatistics statistics;
1002}
1003);
1004
1005impl DataPageHeader {
1006    // reader that skips decoding page statistics
1007    fn read_thrift_without_stats<'a, R>(prot: &mut R) -> Result<Self>
1008    where
1009        R: ThriftCompactInputProtocol<'a>,
1010    {
1011        let mut num_values: Option<i32> = None;
1012        let mut encoding: Option<Encoding> = None;
1013        let mut definition_level_encoding: Option<Encoding> = None;
1014        let mut repetition_level_encoding: Option<Encoding> = None;
1015        let statistics: Option<PageStatistics> = None;
1016        let mut last_field_id = 0i16;
1017        loop {
1018            let field_ident = prot.read_field_begin(last_field_id)?;
1019            if field_ident.field_type == FieldType::Stop {
1020                break;
1021            }
1022            match field_ident.id {
1023                1 => {
1024                    let val = i32::read_thrift(&mut *prot)?;
1025                    num_values = Some(val);
1026                }
1027                2 => {
1028                    let val = Encoding::read_thrift(&mut *prot)?;
1029                    encoding = Some(val);
1030                }
1031                3 => {
1032                    let val = Encoding::read_thrift(&mut *prot)?;
1033                    definition_level_encoding = Some(val);
1034                }
1035                4 => {
1036                    let val = Encoding::read_thrift(&mut *prot)?;
1037                    repetition_level_encoding = Some(val);
1038                }
1039                _ => {
1040                    prot.skip(field_ident.field_type)?;
1041                }
1042            }
1043            last_field_id = field_ident.id;
1044        }
1045        let Some(num_values) = num_values else {
1046            return Err(general_err!("Required field num_values is missing"));
1047        };
1048        let Some(encoding) = encoding else {
1049            return Err(general_err!("Required field encoding is missing"));
1050        };
1051        let Some(definition_level_encoding) = definition_level_encoding else {
1052            return Err(general_err!(
1053                "Required field definition_level_encoding is missing"
1054            ));
1055        };
1056        let Some(repetition_level_encoding) = repetition_level_encoding else {
1057            return Err(general_err!(
1058                "Required field repetition_level_encoding is missing"
1059            ));
1060        };
1061        Ok(Self {
1062            num_values,
1063            encoding,
1064            definition_level_encoding,
1065            repetition_level_encoding,
1066            statistics,
1067        })
1068    }
1069}
1070
1071thrift_struct!(
1072pub(crate) struct DataPageHeaderV2 {
1073  1: required i32 num_values
1074  2: required i32 num_nulls
1075  3: required i32 num_rows
1076  4: required Encoding encoding
1077  5: required i32 definition_levels_byte_length;
1078  6: required i32 repetition_levels_byte_length;
1079  7: optional bool is_compressed = true;
1080  8: optional PageStatistics statistics;
1081}
1082);
1083
1084impl DataPageHeaderV2 {
1085    // reader that skips decoding page statistics
1086    fn read_thrift_without_stats<'a, R>(prot: &mut R) -> Result<Self>
1087    where
1088        R: ThriftCompactInputProtocol<'a>,
1089    {
1090        let mut num_values: Option<i32> = None;
1091        let mut num_nulls: Option<i32> = None;
1092        let mut num_rows: Option<i32> = None;
1093        let mut encoding: Option<Encoding> = None;
1094        let mut definition_levels_byte_length: Option<i32> = None;
1095        let mut repetition_levels_byte_length: Option<i32> = None;
1096        let mut is_compressed: Option<bool> = None;
1097        let statistics: Option<PageStatistics> = None;
1098        let mut last_field_id = 0i16;
1099        loop {
1100            let field_ident = prot.read_field_begin(last_field_id)?;
1101            if field_ident.field_type == FieldType::Stop {
1102                break;
1103            }
1104            match field_ident.id {
1105                1 => {
1106                    let val = i32::read_thrift(&mut *prot)?;
1107                    num_values = Some(val);
1108                }
1109                2 => {
1110                    let val = i32::read_thrift(&mut *prot)?;
1111                    num_nulls = Some(val);
1112                }
1113                3 => {
1114                    let val = i32::read_thrift(&mut *prot)?;
1115                    num_rows = Some(val);
1116                }
1117                4 => {
1118                    let val = Encoding::read_thrift(&mut *prot)?;
1119                    encoding = Some(val);
1120                }
1121                5 => {
1122                    let val = i32::read_thrift(&mut *prot)?;
1123                    definition_levels_byte_length = Some(val);
1124                }
1125                6 => {
1126                    let val = i32::read_thrift(&mut *prot)?;
1127                    repetition_levels_byte_length = Some(val);
1128                }
1129                7 => {
1130                    is_compressed = Some(field_ident.bool_val()?);
1131                }
1132                _ => {
1133                    prot.skip(field_ident.field_type)?;
1134                }
1135            }
1136            last_field_id = field_ident.id;
1137        }
1138        let Some(num_values) = num_values else {
1139            return Err(general_err!("Required field num_values is missing"));
1140        };
1141        let Some(num_nulls) = num_nulls else {
1142            return Err(general_err!("Required field num_nulls is missing"));
1143        };
1144        let Some(num_rows) = num_rows else {
1145            return Err(general_err!("Required field num_rows is missing"));
1146        };
1147        let Some(encoding) = encoding else {
1148            return Err(general_err!("Required field encoding is missing"));
1149        };
1150        let Some(definition_levels_byte_length) = definition_levels_byte_length else {
1151            return Err(general_err!(
1152                "Required field definition_levels_byte_length is missing"
1153            ));
1154        };
1155        let Some(repetition_levels_byte_length) = repetition_levels_byte_length else {
1156            return Err(general_err!(
1157                "Required field repetition_levels_byte_length is missing"
1158            ));
1159        };
1160        Ok(Self {
1161            num_values,
1162            num_nulls,
1163            num_rows,
1164            encoding,
1165            definition_levels_byte_length,
1166            repetition_levels_byte_length,
1167            is_compressed,
1168            statistics,
1169        })
1170    }
1171}
1172
1173thrift_struct!(
1174pub(crate) struct PageHeader {
1175  /// the type of the page: indicates which of the *_header fields is set
1176  1: required PageType r#type
1177
1178  /// Uncompressed page size in bytes (not including this header)
1179  2: required i32 uncompressed_page_size
1180
1181  /// Compressed (and potentially encrypted) page size in bytes, not including this header
1182  3: required i32 compressed_page_size
1183
1184  /// The 32-bit CRC checksum for the page, to be calculated as follows:
1185  4: optional i32 crc
1186
1187  // Headers for page specific data.  One only will be set.
1188  5: optional DataPageHeader data_page_header;
1189  6: optional IndexPageHeader index_page_header;
1190  7: optional DictionaryPageHeader dictionary_page_header;
1191  8: optional DataPageHeaderV2 data_page_header_v2;
1192}
1193);
1194
1195impl PageHeader {
1196    // reader that skips reading page statistics. obtained by running
1197    // `cargo expand -p parquet --all-features --lib file::metadata::thrift`
1198    // and modifying the impl of `read_thrift`
1199    pub(crate) fn read_thrift_without_stats<'a, R>(prot: &mut R) -> Result<Self>
1200    where
1201        R: ThriftCompactInputProtocol<'a>,
1202    {
1203        let mut type_: Option<PageType> = None;
1204        let mut uncompressed_page_size: Option<i32> = None;
1205        let mut compressed_page_size: Option<i32> = None;
1206        let mut crc: Option<i32> = None;
1207        let mut data_page_header: Option<DataPageHeader> = None;
1208        let mut index_page_header: Option<IndexPageHeader> = None;
1209        let mut dictionary_page_header: Option<DictionaryPageHeader> = None;
1210        let mut data_page_header_v2: Option<DataPageHeaderV2> = None;
1211        let mut last_field_id = 0i16;
1212        loop {
1213            let field_ident = prot.read_field_begin(last_field_id)?;
1214            if field_ident.field_type == FieldType::Stop {
1215                break;
1216            }
1217            match field_ident.id {
1218                1 => {
1219                    let val = PageType::read_thrift(&mut *prot)?;
1220                    type_ = Some(val);
1221                }
1222                2 => {
1223                    let val = i32::read_thrift(&mut *prot)?;
1224                    uncompressed_page_size = Some(val);
1225                }
1226                3 => {
1227                    let val = i32::read_thrift(&mut *prot)?;
1228                    compressed_page_size = Some(val);
1229                }
1230                4 => {
1231                    let val = i32::read_thrift(&mut *prot)?;
1232                    crc = Some(val);
1233                }
1234                5 => {
1235                    let val = DataPageHeader::read_thrift_without_stats(&mut *prot)?;
1236                    data_page_header = Some(val);
1237                }
1238                6 => {
1239                    let val = IndexPageHeader::read_thrift(&mut *prot)?;
1240                    index_page_header = Some(val);
1241                }
1242                7 => {
1243                    let val = DictionaryPageHeader::read_thrift(&mut *prot)?;
1244                    dictionary_page_header = Some(val);
1245                }
1246                8 => {
1247                    let val = DataPageHeaderV2::read_thrift_without_stats(&mut *prot)?;
1248                    data_page_header_v2 = Some(val);
1249                }
1250                _ => {
1251                    prot.skip(field_ident.field_type)?;
1252                }
1253            }
1254            last_field_id = field_ident.id;
1255        }
1256        let Some(type_) = type_ else {
1257            return Err(general_err!("Required field type_ is missing"));
1258        };
1259        let Some(uncompressed_page_size) = uncompressed_page_size else {
1260            return Err(general_err!(
1261                "Required field uncompressed_page_size is missing"
1262            ));
1263        };
1264        let Some(compressed_page_size) = compressed_page_size else {
1265            return Err(general_err!(
1266                "Required field compressed_page_size is missing"
1267            ));
1268        };
1269        Ok(Self {
1270            r#type: type_,
1271            uncompressed_page_size,
1272            compressed_page_size,
1273            crc,
1274            data_page_header,
1275            index_page_header,
1276            dictionary_page_header,
1277            data_page_header_v2,
1278        })
1279    }
1280}
1281
1282/////////////////////////////////////////////////
1283// helper functions for writing file meta data
1284
1285#[cfg(feature = "encryption")]
1286fn should_write_column_stats(column_chunk: &ColumnChunkMetaData) -> bool {
1287    // If there is encrypted column metadata present,
1288    // the column is encrypted with a different key to the footer or a plaintext footer is used,
1289    // so the statistics are sensitive and shouldn't be written.
1290    column_chunk.encrypted_column_metadata.is_none()
1291}
1292
1293#[cfg(not(feature = "encryption"))]
1294fn should_write_column_stats(_column_chunk: &ColumnChunkMetaData) -> bool {
1295    true
1296}
1297
1298// serialize the bits of the column chunk needed for a thrift ColumnMetaData
1299// struct ColumnMetaData {
1300//   1: required Type type
1301//   2: required list<Encoding> encodings
1302//   3: required list<string> path_in_schema
1303//   4: required CompressionCodec codec
1304//   5: required i64 num_values
1305//   6: required i64 total_uncompressed_size
1306//   7: required i64 total_compressed_size
1307//   8: optional list<KeyValue> key_value_metadata
1308//   9: required i64 data_page_offset
1309//   10: optional i64 index_page_offset
1310//   11: optional i64 dictionary_page_offset
1311//   12: optional Statistics statistics;
1312//   13: optional list<PageEncodingStats> encoding_stats;
1313//   14: optional i64 bloom_filter_offset;
1314//   15: optional i32 bloom_filter_length;
1315//   16: optional SizeStatistics size_statistics;
1316//   17: optional GeospatialStatistics geospatial_statistics;
1317// }
1318pub(super) fn serialize_column_meta_data<W: Write>(
1319    column_chunk: &ColumnChunkMetaData,
1320    w: &mut ThriftCompactOutputProtocol<W>,
1321) -> Result<()> {
1322    use crate::file::statistics::page_stats_to_thrift;
1323
1324    column_chunk.column_type().write_thrift_field(w, 1, 0)?;
1325    column_chunk
1326        .encodings()
1327        .collect::<Vec<_>>()
1328        .write_thrift_field(w, 2, 1)?;
1329    if w.write_path_in_schema() {
1330        let path = column_chunk.column_descr.path().parts();
1331        let path: Vec<&str> = path.iter().map(|v| v.as_str()).collect();
1332        path.write_thrift_field(w, 3, 2)?;
1333        column_chunk.compression.write_thrift_field(w, 4, 3)?;
1334    } else {
1335        column_chunk.compression.write_thrift_field(w, 4, 2)?;
1336    }
1337
1338    column_chunk.num_values.write_thrift_field(w, 5, 4)?;
1339    column_chunk
1340        .total_uncompressed_size
1341        .write_thrift_field(w, 6, 5)?;
1342    column_chunk
1343        .total_compressed_size
1344        .write_thrift_field(w, 7, 6)?;
1345    // no key_value_metadata here
1346    let mut last_field_id = column_chunk.data_page_offset.write_thrift_field(w, 9, 7)?;
1347    if let Some(index_page_offset) = column_chunk.index_page_offset {
1348        last_field_id = index_page_offset.write_thrift_field(w, 10, last_field_id)?;
1349    }
1350    if let Some(dictionary_page_offset) = column_chunk.dictionary_page_offset {
1351        last_field_id = dictionary_page_offset.write_thrift_field(w, 11, last_field_id)?;
1352    }
1353
1354    if should_write_column_stats(column_chunk) {
1355        // PageStatistics is the same as thrift Statistics, but writable
1356        let stats = page_stats_to_thrift(column_chunk.statistics());
1357        if let Some(stats) = stats {
1358            last_field_id = stats.write_thrift_field(w, 12, last_field_id)?;
1359        }
1360        if let Some(page_encoding_stats) = column_chunk.page_encoding_stats() {
1361            last_field_id = page_encoding_stats.write_thrift_field(w, 13, last_field_id)?;
1362        }
1363        if let Some(bloom_filter_offset) = column_chunk.bloom_filter_offset {
1364            last_field_id = bloom_filter_offset.write_thrift_field(w, 14, last_field_id)?;
1365        }
1366        if let Some(bloom_filter_length) = column_chunk.bloom_filter_length {
1367            last_field_id = bloom_filter_length.write_thrift_field(w, 15, last_field_id)?;
1368        }
1369
1370        // SizeStatistics
1371        let size_stats = if column_chunk.unencoded_byte_array_data_bytes.is_some()
1372            || column_chunk.repetition_level_histogram.is_some()
1373            || column_chunk.definition_level_histogram.is_some()
1374        {
1375            let repetition_level_histogram = column_chunk
1376                .repetition_level_histogram()
1377                .map(|hist| hist.clone().into_inner());
1378
1379            let definition_level_histogram = column_chunk
1380                .definition_level_histogram()
1381                .map(|hist| hist.clone().into_inner());
1382
1383            Some(SizeStatistics {
1384                unencoded_byte_array_data_bytes: column_chunk.unencoded_byte_array_data_bytes,
1385                repetition_level_histogram,
1386                definition_level_histogram,
1387            })
1388        } else {
1389            None
1390        };
1391        if let Some(size_stats) = size_stats {
1392            last_field_id = size_stats.write_thrift_field(w, 16, last_field_id)?;
1393        }
1394
1395        if let Some(geo_stats) = column_chunk.geo_statistics() {
1396            geo_stats.write_thrift_field(w, 17, last_field_id)?;
1397        }
1398    }
1399
1400    w.write_struct_end()
1401}
1402
1403// temp struct used for writing
1404pub(super) struct FileMeta<'a> {
1405    pub(super) file_metadata: &'a crate::file::metadata::FileMetaData,
1406    pub(super) row_groups: &'a Vec<RowGroupMetaData>,
1407    // If true, then write the `path_in_schema` field in the ColumnMetaData struct.
1408    pub(super) write_path_in_schema: bool,
1409}
1410
1411// struct FileMetaData {
1412//   1: required i32 version
1413//   2: required list<SchemaElement> schema;
1414//   3: required i64 num_rows
1415//   4: required list<RowGroup> row_groups
1416//   5: optional list<KeyValue> key_value_metadata
1417//   6: optional string created_by
1418//   7: optional list<ColumnOrder> column_orders;
1419//   8: optional EncryptionAlgorithm encryption_algorithm
1420//   9: optional binary footer_signing_key_metadata
1421// }
1422impl WriteThrift for FileMeta<'_> {
1423    const ELEMENT_TYPE: ElementType = ElementType::Struct;
1424
1425    // needed for last_field_id w/o encryption
1426    #[cfg_attr(not(feature = "encryption"), expect(unused_assignments))]
1427    fn write_thrift<W: Write>(&self, writer: &mut ThriftCompactOutputProtocol<W>) -> Result<()> {
1428        writer.set_write_path_in_schema(self.write_path_in_schema);
1429        // only write ordinal if all values will fit in an i16
1430        writer.set_write_row_group_ordinal(i16::try_from(self.row_groups.len()).is_ok());
1431
1432        self.file_metadata
1433            .version
1434            .write_thrift_field(writer, 1, 0)?;
1435
1436        // field 2 is schema. do depth-first traversal of tree, converting to SchemaElement and
1437        // writing along the way.
1438        let root = self.file_metadata.schema_descr().root_schema_ptr();
1439        let schema_len = num_nodes(&root)?;
1440        writer.write_field_begin(FieldType::List, 2, 1)?;
1441        writer.write_list_begin(ElementType::Struct, schema_len)?;
1442        // recursively write Type nodes as SchemaElements
1443        write_schema(&root, writer)?;
1444
1445        self.file_metadata
1446            .num_rows
1447            .write_thrift_field(writer, 3, 2)?;
1448
1449        // this will call RowGroupMetaData::write_thrift
1450        let mut last_field_id = self.row_groups.write_thrift_field(writer, 4, 3)?;
1451
1452        if let Some(kv_metadata) = self.file_metadata.key_value_metadata() {
1453            last_field_id = kv_metadata.write_thrift_field(writer, 5, last_field_id)?;
1454        }
1455        if let Some(created_by) = self.file_metadata.created_by() {
1456            last_field_id = created_by.write_thrift_field(writer, 6, last_field_id)?;
1457        }
1458        if let Some(column_orders) = self.file_metadata.column_orders() {
1459            last_field_id = column_orders.write_thrift_field(writer, 7, last_field_id)?;
1460        }
1461        #[cfg(feature = "encryption")]
1462        if let Some(algo) = self.file_metadata.encryption_algorithm.as_ref() {
1463            last_field_id = algo.write_thrift_field(writer, 8, last_field_id)?;
1464        }
1465        #[cfg(feature = "encryption")]
1466        if let Some(key) = self.file_metadata.footer_signing_key_metadata.as_ref() {
1467            key.as_slice()
1468                .write_thrift_field(writer, 9, last_field_id)?;
1469        }
1470
1471        writer.write_struct_end()
1472    }
1473}
1474
1475fn write_schema<W: Write>(
1476    schema: &TypePtr,
1477    writer: &mut ThriftCompactOutputProtocol<W>,
1478) -> Result<()> {
1479    if !schema.is_group() {
1480        return Err(general_err!("Root schema must be Group type"));
1481    }
1482    write_schema_helper(schema, writer)
1483}
1484
1485fn write_schema_helper<W: Write>(
1486    node: &TypePtr,
1487    writer: &mut ThriftCompactOutputProtocol<W>,
1488) -> Result<()> {
1489    match node.as_ref() {
1490        crate::schema::types::Type::PrimitiveType {
1491            basic_info,
1492            physical_type,
1493            type_length,
1494            scale,
1495            precision,
1496        } => {
1497            let element = SchemaElement {
1498                r#type: Some(*physical_type),
1499                type_length: if *type_length >= 0 {
1500                    Some(*type_length)
1501                } else {
1502                    None
1503                },
1504                repetition_type: Some(basic_info.repetition()),
1505                name: basic_info.name(),
1506                num_children: None,
1507                converted_type: match basic_info.converted_type() {
1508                    ConvertedType::NONE => None,
1509                    other => Some(other),
1510                },
1511                scale: if *scale >= 0 { Some(*scale) } else { None },
1512                precision: if *precision >= 0 {
1513                    Some(*precision)
1514                } else {
1515                    None
1516                },
1517                field_id: if basic_info.has_id() {
1518                    Some(basic_info.id())
1519                } else {
1520                    None
1521                },
1522                logical_type: basic_info.logical_type_ref().cloned(),
1523            };
1524            element.write_thrift(writer)
1525        }
1526        crate::schema::types::Type::GroupType { basic_info, fields } => {
1527            let repetition = if basic_info.has_repetition() {
1528                Some(basic_info.repetition())
1529            } else {
1530                None
1531            };
1532
1533            let element = SchemaElement {
1534                r#type: None,
1535                type_length: None,
1536                repetition_type: repetition,
1537                name: basic_info.name(),
1538                num_children: Some(fields.len().try_into()?),
1539                converted_type: match basic_info.converted_type() {
1540                    ConvertedType::NONE => None,
1541                    other => Some(other),
1542                },
1543                scale: None,
1544                precision: None,
1545                field_id: if basic_info.has_id() {
1546                    Some(basic_info.id())
1547                } else {
1548                    None
1549                },
1550                logical_type: basic_info.logical_type_ref().cloned(),
1551            };
1552
1553            element.write_thrift(writer)?;
1554
1555            // Add child elements for a group
1556            for field in fields {
1557                write_schema_helper(field, writer)?;
1558            }
1559            Ok(())
1560        }
1561    }
1562}
1563
1564// struct RowGroup {
1565//   1: required list<ColumnChunk> columns
1566//   2: required i64 total_byte_size
1567//   3: required i64 num_rows
1568//   4: optional list<SortingColumn> sorting_columns
1569//   5: optional i64 file_offset
1570//   6: optional i64 total_compressed_size
1571//   7: optional i16 ordinal
1572// }
1573impl WriteThrift for RowGroupMetaData {
1574    const ELEMENT_TYPE: ElementType = ElementType::Struct;
1575
1576    fn write_thrift<W: Write>(&self, writer: &mut ThriftCompactOutputProtocol<W>) -> Result<()> {
1577        // this will call ColumnChunkMetaData::write_thrift
1578        self.columns.write_thrift_field(writer, 1, 0)?;
1579        self.total_byte_size.write_thrift_field(writer, 2, 1)?;
1580        let mut last_field_id = self.num_rows.write_thrift_field(writer, 3, 2)?;
1581        if let Some(sorting_columns) = self.sorting_columns() {
1582            last_field_id = sorting_columns.write_thrift_field(writer, 4, last_field_id)?;
1583        }
1584        if let Some(file_offset) = self.file_offset() {
1585            last_field_id = file_offset.write_thrift_field(writer, 5, last_field_id)?;
1586        }
1587        // this is optional, but we'll always write it
1588        last_field_id = self
1589            .compressed_size()
1590            .write_thrift_field(writer, 6, last_field_id)?;
1591
1592        // write ordinal if it will fit in an i16
1593        if writer.write_row_group_ordinal()
1594            && let Some(ordinal) = self.ordinal()
1595            && let Ok(ordinal) = i16::try_from(ordinal)
1596        {
1597            ordinal.write_thrift_field(writer, 7, last_field_id)?;
1598        }
1599        writer.write_struct_end()
1600    }
1601}
1602
1603// struct ColumnChunk {
1604//   1: optional string file_path
1605//   2: required i64 file_offset = 0
1606//   3: optional ColumnMetaData meta_data
1607//   4: optional i64 offset_index_offset
1608//   5: optional i32 offset_index_length
1609//   6: optional i64 column_index_offset
1610//   7: optional i32 column_index_length
1611//   8: optional ColumnCryptoMetaData crypto_metadata
1612//   9: optional binary encrypted_column_metadata
1613// }
1614impl WriteThrift for ColumnChunkMetaData {
1615    const ELEMENT_TYPE: ElementType = ElementType::Struct;
1616
1617    #[cfg_attr(not(feature = "encryption"), expect(unused_assignments))]
1618    fn write_thrift<W: Write>(&self, writer: &mut ThriftCompactOutputProtocol<W>) -> Result<()> {
1619        let mut last_field_id = 0i16;
1620        if let Some(file_path) = self.file_path() {
1621            last_field_id = file_path.write_thrift_field(writer, 1, last_field_id)?;
1622        }
1623        last_field_id = self
1624            .file_offset()
1625            .write_thrift_field(writer, 2, last_field_id)?;
1626
1627        #[cfg(feature = "encryption")]
1628        let write_meta_data =
1629            self.encrypted_column_metadata.is_none() || self.plaintext_footer_mode;
1630        #[cfg(not(feature = "encryption"))]
1631        let write_meta_data = true;
1632
1633        // When the footer is encrypted and encrypted_column_metadata is present,
1634        // skip writing the plaintext meta_data field to reduce footer size.
1635        // When the footer is plaintext (plaintext_footer_mode=true), we still write
1636        // meta_data for backward compatibility with readers that expect it, but with
1637        // sensitive fields (statistics, bloom filter info, etc.) stripped out.
1638        if write_meta_data {
1639            writer.write_field_begin(FieldType::Struct, 3, last_field_id)?;
1640            serialize_column_meta_data(self, writer)?;
1641            last_field_id = 3;
1642        }
1643
1644        if let Some(offset_idx_off) = self.offset_index_offset() {
1645            last_field_id = offset_idx_off.write_thrift_field(writer, 4, last_field_id)?;
1646        }
1647        if let Some(offset_idx_len) = self.offset_index_length() {
1648            last_field_id = offset_idx_len.write_thrift_field(writer, 5, last_field_id)?;
1649        }
1650        if let Some(column_idx_off) = self.column_index_offset() {
1651            last_field_id = column_idx_off.write_thrift_field(writer, 6, last_field_id)?;
1652        }
1653        if let Some(column_idx_len) = self.column_index_length() {
1654            last_field_id = column_idx_len.write_thrift_field(writer, 7, last_field_id)?;
1655        }
1656        #[cfg(feature = "encryption")]
1657        {
1658            if let Some(crypto_metadata) = self.crypto_metadata() {
1659                last_field_id = crypto_metadata.write_thrift_field(writer, 8, last_field_id)?;
1660            }
1661            if let Some(encrypted_meta) = self.encrypted_column_metadata.as_ref() {
1662                encrypted_meta
1663                    .as_slice()
1664                    .write_thrift_field(writer, 9, last_field_id)?;
1665            }
1666        }
1667
1668        writer.write_struct_end()
1669    }
1670}
1671
1672// struct GeospatialStatistics {
1673//   1: optional BoundingBox bbox;
1674//   2: optional list<i32> geospatial_types;
1675// }
1676impl WriteThrift for crate::geospatial::statistics::GeospatialStatistics {
1677    const ELEMENT_TYPE: ElementType = ElementType::Struct;
1678
1679    fn write_thrift<W: Write>(&self, writer: &mut ThriftCompactOutputProtocol<W>) -> Result<()> {
1680        let mut last_field_id = 0i16;
1681        if let Some(bbox) = self.bounding_box() {
1682            last_field_id = bbox.write_thrift_field(writer, 1, last_field_id)?;
1683        }
1684        if let Some(geo_types) = self.geospatial_types() {
1685            geo_types.write_thrift_field(writer, 2, last_field_id)?;
1686        }
1687
1688        writer.write_struct_end()
1689    }
1690}
1691
1692// macro cannot handle qualified names
1693use crate::geospatial::statistics::GeospatialStatistics as RustGeospatialStatistics;
1694write_thrift_field!(RustGeospatialStatistics, FieldType::Struct);
1695
1696// struct BoundingBox {
1697//   1: required double xmin;
1698//   2: required double xmax;
1699//   3: required double ymin;
1700//   4: required double ymax;
1701//   5: optional double zmin;
1702//   6: optional double zmax;
1703//   7: optional double mmin;
1704//   8: optional double mmax;
1705// }
1706impl WriteThrift for crate::geospatial::bounding_box::BoundingBox {
1707    const ELEMENT_TYPE: ElementType = ElementType::Struct;
1708
1709    fn write_thrift<W: Write>(&self, writer: &mut ThriftCompactOutputProtocol<W>) -> Result<()> {
1710        self.get_xmin().write_thrift_field(writer, 1, 0)?;
1711        self.get_xmax().write_thrift_field(writer, 2, 1)?;
1712        self.get_ymin().write_thrift_field(writer, 3, 2)?;
1713        let mut last_field_id = self.get_ymax().write_thrift_field(writer, 4, 3)?;
1714
1715        if let Some(zmin) = self.get_zmin() {
1716            last_field_id = zmin.write_thrift_field(writer, 5, last_field_id)?;
1717        }
1718        if let Some(zmax) = self.get_zmax() {
1719            last_field_id = zmax.write_thrift_field(writer, 6, last_field_id)?;
1720        }
1721        if let Some(mmin) = self.get_mmin() {
1722            last_field_id = mmin.write_thrift_field(writer, 7, last_field_id)?;
1723        }
1724        if let Some(mmax) = self.get_mmax() {
1725            mmax.write_thrift_field(writer, 8, last_field_id)?;
1726        }
1727
1728        writer.write_struct_end()
1729    }
1730}
1731
1732// macro cannot handle qualified names
1733use crate::geospatial::bounding_box::BoundingBox as RustBoundingBox;
1734write_thrift_field!(RustBoundingBox, FieldType::Struct);
1735
1736#[cfg(test)]
1737pub(crate) mod tests {
1738    use crate::basic::{Encoding, PageType, Type as PhysicalType};
1739    use crate::errors::Result;
1740    use crate::file::metadata::thrift::{
1741        BoundingBox, DataPageHeaderV2, DictionaryPageHeader, PageHeader, SchemaElement,
1742        write_schema,
1743    };
1744    use crate::file::metadata::{ColumnChunkMetaData, ParquetMetaDataOptions, RowGroupMetaData};
1745    use crate::parquet_thrift::tests::test_roundtrip;
1746    use crate::parquet_thrift::{
1747        ElementType, ThriftCompactOutputProtocol, ThriftSliceInputProtocol, WriteThrift,
1748        read_thrift_vec,
1749    };
1750    use crate::schema::types::{
1751        ColumnDescriptor, ColumnPath, SchemaDescriptor, TypePtr, num_nodes,
1752        parquet_schema_from_array,
1753    };
1754    use std::sync::Arc;
1755
1756    // for testing. decode thrift encoded RowGroup
1757    pub(crate) fn read_row_group(
1758        buf: &[u8],
1759        schema_descr: Arc<SchemaDescriptor>,
1760    ) -> Result<RowGroupMetaData> {
1761        let mut reader = ThriftSliceInputProtocol::new(buf);
1762        crate::file::metadata::thrift::read_row_group(&mut reader, &schema_descr, None)
1763    }
1764
1765    pub(crate) fn read_column_chunk(
1766        buf: &[u8],
1767        column_descr: Arc<ColumnDescriptor>,
1768    ) -> Result<ColumnChunkMetaData> {
1769        read_column_chunk_with_options(buf, column_descr, None)
1770    }
1771
1772    pub(crate) fn read_column_chunk_with_options(
1773        buf: &[u8],
1774        column_descr: Arc<ColumnDescriptor>,
1775        options: Option<&ParquetMetaDataOptions>,
1776    ) -> Result<ColumnChunkMetaData> {
1777        let mut reader = ThriftSliceInputProtocol::new(buf);
1778        crate::file::metadata::thrift::read_column_chunk(&mut reader, &column_descr, 0, options)
1779    }
1780
1781    pub(crate) fn roundtrip_schema(schema: TypePtr) -> Result<TypePtr> {
1782        let num_nodes = num_nodes(&schema)?;
1783        let mut buf = Vec::new();
1784        let mut writer = ThriftCompactOutputProtocol::new(&mut buf);
1785
1786        // kick off writing list
1787        writer.write_list_begin(ElementType::Struct, num_nodes)?;
1788
1789        // write SchemaElements
1790        write_schema(&schema, &mut writer)?;
1791
1792        let mut prot = ThriftSliceInputProtocol::new(&buf);
1793        let se: Vec<SchemaElement> = read_thrift_vec(&mut prot)?;
1794        parquet_schema_from_array(se)
1795    }
1796
1797    pub(crate) fn schema_to_buf(schema: &TypePtr) -> Result<Vec<u8>> {
1798        let num_nodes = num_nodes(schema)?;
1799        let mut buf = Vec::new();
1800        let mut writer = ThriftCompactOutputProtocol::new(&mut buf);
1801
1802        // kick off writing list
1803        writer.write_list_begin(ElementType::Struct, num_nodes)?;
1804
1805        // write SchemaElements
1806        write_schema(schema, &mut writer)?;
1807        Ok(buf)
1808    }
1809
1810    pub(crate) fn buf_to_schema_list(buf: &mut Vec<u8>) -> Result<Vec<SchemaElement<'_>>> {
1811        let mut prot = ThriftSliceInputProtocol::new(buf.as_mut_slice());
1812        read_thrift_vec(&mut prot)
1813    }
1814
1815    fn thrift_bytes<T: WriteThrift>(value: &T) -> Vec<u8> {
1816        let mut buf = Vec::new();
1817        let mut writer = ThriftCompactOutputProtocol::new(&mut buf);
1818        value.write_thrift(&mut writer).unwrap();
1819        buf
1820    }
1821
1822    fn change_false_bool_field_to_i32(buf: &mut [u8]) {
1823        let pos = buf
1824            .iter()
1825            .rposition(|byte| *byte == 0x12)
1826            .expect("expected BOOL_FALSE field header byte");
1827        buf[pos] = 0x15;
1828    }
1829
1830    fn assert_malformed_bool_error(err: crate::errors::ParquetError) {
1831        let msg = err.to_string();
1832        assert!(
1833            msg.contains("Unexpected struct field type"),
1834            "unexpected error message: {msg}"
1835        );
1836    }
1837
1838    #[test]
1839    fn test_bounding_box_roundtrip() {
1840        test_roundtrip(BoundingBox {
1841            xmin: 0.1.into(),
1842            xmax: 10.3.into(),
1843            ymin: 0.001.into(),
1844            ymax: 128.5.into(),
1845            zmin: None,
1846            zmax: None,
1847            mmin: None,
1848            mmax: None,
1849        });
1850
1851        test_roundtrip(BoundingBox {
1852            xmin: 0.1.into(),
1853            xmax: 10.3.into(),
1854            ymin: 0.001.into(),
1855            ymax: 128.5.into(),
1856            zmin: Some(11.0.into()),
1857            zmax: Some(1300.0.into()),
1858            mmin: None,
1859            mmax: None,
1860        });
1861
1862        test_roundtrip(BoundingBox {
1863            xmin: 0.1.into(),
1864            xmax: 10.3.into(),
1865            ymin: 0.001.into(),
1866            ymax: 128.5.into(),
1867            zmin: Some(11.0.into()),
1868            zmax: Some(1300.0.into()),
1869            mmin: Some(3.7.into()),
1870            mmax: Some(42.0.into()),
1871        });
1872    }
1873
1874    #[test]
1875    fn test_convert_stats_preserves_missing_null_count() {
1876        let primitive =
1877            crate::schema::types::Type::primitive_type_builder("col", PhysicalType::INT32)
1878                .build()
1879                .unwrap();
1880        let column_descr = Arc::new(ColumnDescriptor::new(
1881            Arc::new(primitive),
1882            0,
1883            0,
1884            ColumnPath::new(vec![]),
1885        ));
1886
1887        let none_null_count = super::Statistics {
1888            max: None,
1889            min: None,
1890            null_count: None,
1891            distinct_count: None,
1892            max_value: None,
1893            min_value: None,
1894            is_max_value_exact: None,
1895            is_min_value_exact: None,
1896            nan_count: None,
1897        };
1898        let decoded_none = super::convert_stats(&column_descr, Some(none_null_count))
1899            .unwrap()
1900            .unwrap();
1901        assert_eq!(decoded_none.null_count_opt(), None);
1902
1903        let zero_null_count = super::Statistics {
1904            max: None,
1905            min: None,
1906            null_count: Some(0),
1907            distinct_count: None,
1908            max_value: None,
1909            min_value: None,
1910            is_max_value_exact: None,
1911            is_min_value_exact: None,
1912            nan_count: None,
1913        };
1914        let decoded_zero = super::convert_stats(&column_descr, Some(zero_null_count))
1915            .unwrap()
1916            .unwrap();
1917        assert_eq!(decoded_zero.null_count_opt(), Some(0));
1918    }
1919
1920    #[test]
1921    fn test_convert_stats_returns_error_for_overlong_int96_statistics() {
1922        let primitive =
1923            crate::schema::types::Type::primitive_type_builder("col", PhysicalType::INT96)
1924                .build()
1925                .unwrap();
1926        let column_descr = Arc::new(ColumnDescriptor::new(
1927            Arc::new(primitive),
1928            0,
1929            0,
1930            ColumnPath::new(vec![]),
1931        ));
1932        let invalid = (0..13).collect::<Vec<_>>();
1933
1934        let make_stats = |min, max| super::Statistics {
1935            max,
1936            min,
1937            null_count: Some(0),
1938            distinct_count: None,
1939            max_value: None,
1940            min_value: None,
1941            is_max_value_exact: None,
1942            is_min_value_exact: None,
1943            nan_count: None,
1944        };
1945
1946        let err = super::convert_stats(&column_descr, Some(make_stats(Some(&invalid), None)))
1947            .unwrap_err();
1948        assert_eq!(
1949            err.to_string(),
1950            "Parquet error: Incorrect Int96 min statistics"
1951        );
1952
1953        let err = super::convert_stats(&column_descr, Some(make_stats(None, Some(&invalid))))
1954            .unwrap_err();
1955        assert_eq!(
1956            err.to_string(),
1957            "Parquet error: Incorrect Int96 max statistics"
1958        );
1959    }
1960
1961    #[test]
1962    fn malformed_bool_field_returns_error_not_panic() {
1963        let page_header = PageHeader {
1964            r#type: PageType::DICTIONARY_PAGE,
1965            uncompressed_page_size: 1,
1966            compressed_page_size: 1,
1967            crc: None,
1968            data_page_header: None,
1969            index_page_header: None,
1970            dictionary_page_header: Some(DictionaryPageHeader {
1971                num_values: 1,
1972                encoding: Encoding::PLAIN,
1973                is_sorted: Some(false),
1974            }),
1975            data_page_header_v2: None,
1976        };
1977
1978        let mut buf = thrift_bytes(&page_header);
1979        change_false_bool_field_to_i32(&mut buf);
1980
1981        let mut prot = ThriftSliceInputProtocol::new(&buf);
1982        let err = PageHeader::read_thrift_without_stats(&mut prot)
1983            .expect_err("malformed bool field should return an error");
1984        assert_malformed_bool_error(err);
1985    }
1986
1987    #[test]
1988    fn malformed_data_page_v2_bool_field_returns_error_not_panic() {
1989        let data_page_header_v2 = DataPageHeaderV2 {
1990            num_values: 1,
1991            num_nulls: 0,
1992            num_rows: 1,
1993            encoding: Encoding::PLAIN,
1994            definition_levels_byte_length: 0,
1995            repetition_levels_byte_length: 0,
1996            is_compressed: Some(false),
1997            statistics: None,
1998        };
1999
2000        let mut buf = thrift_bytes(&data_page_header_v2);
2001        change_false_bool_field_to_i32(&mut buf);
2002
2003        let mut prot = ThriftSliceInputProtocol::new(&buf);
2004        let err = DataPageHeaderV2::read_thrift_without_stats(&mut prot)
2005            .expect_err("malformed bool field should return an error");
2006        assert_malformed_bool_error(err);
2007    }
2008
2009    /// Round-trip [`crate::file::metadata::ParquetMetaData`] with the given
2010    /// per-row-group ordinals through thrift encode → decode, returning the
2011    /// decoded ordinals. Exercises `ensure_row_group_ordinals`.
2012    fn roundtrip_rg_ordinals(ordinals: &[Option<i32>]) -> Vec<Option<i32>> {
2013        use crate::file::metadata::ParquetMetaDataWriter;
2014        use crate::file::metadata::{FileMetaData, ParquetMetaData, ParquetMetaDataReader};
2015        use crate::schema::types::Type as SchemaType;
2016
2017        let field = SchemaType::primitive_type_builder("c", PhysicalType::INT32)
2018            .build()
2019            .unwrap();
2020        let schema = SchemaType::group_type_builder("schema")
2021            .with_fields(vec![Arc::new(field)])
2022            .build()
2023            .unwrap();
2024        let schema_descr = Arc::new(SchemaDescriptor::new(Arc::new(schema)));
2025
2026        let row_groups = ordinals
2027            .iter()
2028            .map(|ordinal| {
2029                let columns = schema_descr
2030                    .columns()
2031                    .iter()
2032                    .map(|col| ColumnChunkMetaData::builder(col.clone()).build().unwrap())
2033                    .collect();
2034                let mut builder =
2035                    crate::file::metadata::RowGroupMetaData::builder(schema_descr.clone())
2036                        .set_num_rows(10)
2037                        .set_total_byte_size(100)
2038                        .set_column_metadata(columns);
2039                if let Some(ordinal) = ordinal {
2040                    builder = builder.set_ordinal(*ordinal);
2041                }
2042                builder.build().unwrap()
2043            })
2044            .collect();
2045
2046        let file_metadata = FileMetaData::new(
2047            1,
2048            10 * ordinals.len() as i64,
2049            None,
2050            None,
2051            schema_descr,
2052            None,
2053        );
2054        let metadata = ParquetMetaData::new(file_metadata, row_groups);
2055
2056        let mut buffer = Vec::new();
2057        ParquetMetaDataWriter::new(&mut buffer, &metadata)
2058            .finish()
2059            .unwrap();
2060        // strip the 8-byte footer tail (length + magic)
2061        let decoded = ParquetMetaDataReader::decode_metadata(&buffer[..buffer.len() - 8]).unwrap();
2062        decoded.row_groups().iter().map(|rg| rg.ordinal()).collect()
2063    }
2064
2065    /// All row groups carry ordinals: honored as written, even when they do
2066    /// not match file position.
2067    #[test]
2068    fn ordinals_all_present_are_honored() {
2069        assert_eq!(
2070            roundtrip_rg_ordinals(&[Some(5), Some(1), Some(3)]),
2071            vec![Some(5), Some(1), Some(3)],
2072        );
2073    }
2074
2075    /// No row group carries an ordinal: sequential-filled at decode time so
2076    /// downstream consumers behave identically on fresh vs reused metadata.
2077    #[test]
2078    fn ordinals_none_present_are_sequentially_filled() {
2079        assert_eq!(
2080            roundtrip_rg_ordinals(&[None, None, None]),
2081            vec![Some(0), Some(1), Some(2)],
2082        );
2083    }
2084
2085    /// Mixed ordinals (spec-valid; produced by e.g. Go parquet writers):
2086    /// decode succeeds — the pre-#8715 behavior restored by #10381 — and the
2087    /// metadata is left untouched so row numbering fails deterministically
2088    /// rather than producing numbers that depend on row-group selection.
2089    #[test]
2090    fn ordinals_mixed_decode_succeeds_untouched() {
2091        assert_eq!(
2092            roundtrip_rg_ordinals(&[Some(0), None, Some(2)]),
2093            vec![Some(0), None, Some(2)],
2094        );
2095        // first missing, rest present — the exact Go-writer shape from #10381
2096        assert_eq!(
2097            roundtrip_rg_ordinals(&[None, Some(1), Some(2)]),
2098            vec![None, Some(1), Some(2)],
2099        );
2100    }
2101
2102    /// Test row group on-wire size validation.
2103    #[test]
2104    fn row_group_count_exceeding_remaining_input_is_rejected() {
2105        use crate::file::metadata::thrift::parquet_metadata_from_bytes;
2106        use crate::parquet_thrift::{FieldType, WriteThriftField};
2107        use crate::schema::types::Type as SchemaType;
2108
2109        // Minimal valid schema: a group with one INT32 column.
2110        let field = SchemaType::primitive_type_builder("c", PhysicalType::INT32)
2111            .build()
2112            .unwrap();
2113        let schema = SchemaType::group_type_builder("schema")
2114            .with_fields(vec![Arc::new(field)])
2115            .build()
2116            .unwrap();
2117        let root: TypePtr = Arc::new(schema);
2118        let schema_len = num_nodes(&root).unwrap();
2119
2120        let mut buf = Vec::new();
2121        let mut w = ThriftCompactOutputProtocol::new(&mut buf);
2122        // FileMetaData field 1: version.
2123        1i32.write_thrift_field(&mut w, 1, 0).unwrap();
2124        // field 2: schema (required before row groups are read).
2125        w.write_field_begin(FieldType::List, 2, 1).unwrap();
2126        w.write_list_begin(ElementType::Struct, schema_len).unwrap();
2127        write_schema(&root, &mut w).unwrap();
2128        // field 3: num_rows.
2129        0i64.write_thrift_field(&mut w, 3, 2).unwrap();
2130        // field 4: row_groups, declaring 1024 structs with no element bytes
2131        // after the header. Only the struct-stop byte remains.
2132        w.write_field_begin(FieldType::List, 4, 3).unwrap();
2133        w.write_list_begin(ElementType::Struct, 1024).unwrap();
2134        w.write_struct_end().unwrap();
2135
2136        let err = parquet_metadata_from_bytes(&buf, None)
2137            .expect_err("over-declared row-group count must be rejected")
2138            .to_string();
2139        assert!(
2140            err.contains("exceeds remaining input length"),
2141            "expected a remaining-input bound error, got: {err}"
2142        );
2143    }
2144}