1use 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 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
64thrift_struct!(
66pub(crate) struct SchemaElement<'a> {
67 1: optional Type r#type;
69 2: optional i32 type_length;
74 3: optional Repetition repetition_type;
77 4: required string<'a> name;
79 5: optional i32 num_children;
84 6: optional ConvertedType converted_type;
89 7: optional i32 scale
94 8: optional i32 precision
95 9: optional i32 field_id;
98 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 _ => newbb,
168 };
169
170 newbb = match (bb.mmin, bb.mmax) {
171 (Some(mmin), Some(mmax)) => newbb.with_mrange(mmin.into(), mmax.into()),
172 _ => newbb,
174 };
175
176 newbb
177}
178
179fn 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 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 let distinct_count = stats.distinct_count.map(|value| value as u64);
202 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 let old_format = stats.min_value.is_none() && stats.max_value.is_none();
217 let min = if old_format {
219 stats.min
220 } else {
221 stats.min_value
222 };
223 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 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 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
350const 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
359const 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
368fn 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<'a>(
401 prot: &mut ThriftSliceInputProtocol<'a>,
402) -> Result<EncodingMask> {
403 let mut mask = 0i32;
405 let list_ident = prot.read_list_begin()?;
406 validate_list_type(ElementType::Struct, &list_ident)?;
408
409 for _ in 0..list_ident.size {
410 let pes = PageEncodingStats::read_thrift(prot)?;
411 match pes.page_type {
412 PageType::DATA_PAGE | PageType::DATA_PAGE_V2 => mask |= 1 << pes.encoding as i32,
413 _ => {}
414 }
415 }
416 EncodingMask::try_new(mask)
417}
418
419#[expect(clippy::useless_let_if_seq)] fn read_column_metadata<'a>(
423 prot: &mut ThriftSliceInputProtocol<'a>,
424 column: &mut ColumnChunkMetaData,
425 col_index: usize,
426 options: Option<&ParquetMetaDataOptions>,
427) -> Result<u16> {
428 let mut seen_mask = 0u16;
430
431 let mut skip_pes = false;
432 let mut pes_mask = true;
433 let mut skip_col_stats = false;
434 let mut skip_size_stats = false;
435
436 if let Some(opts) = options {
437 skip_pes = opts.skip_encoding_stats(col_index);
438 pes_mask = opts.encoding_stats_as_mask();
439 skip_col_stats = opts.skip_column_stats(col_index);
440 skip_size_stats = opts.skip_size_stats(col_index);
441 }
442
443 let column_descr = &column.column_descr;
463
464 let mut last_field_id = 0i16;
465 loop {
466 let field_ident = prot.read_field_begin(last_field_id)?;
467 if field_ident.field_type == FieldType::Stop {
468 break;
469 }
470 match field_ident.id {
471 1 => {
473 Type::read_thrift(&mut *prot)?;
475 seen_mask |= COL_META_TYPE;
476 }
477 2 => {
478 column.encodings = EncodingMask::read_thrift(&mut *prot)?;
479 seen_mask |= COL_META_ENCODINGS;
480 }
481 4 => {
483 column.compression = CompressionCodec::read_thrift(&mut *prot)?;
484 seen_mask |= COL_META_CODEC;
485 }
486 5 => {
487 column.num_values = i64::read_thrift(&mut *prot)?;
488 seen_mask |= COL_META_NUM_VALUES;
489 }
490 6 => {
491 column.total_uncompressed_size = i64::read_thrift(&mut *prot)?;
492 seen_mask |= COL_META_TOTAL_UNCOMP_SZ;
493 }
494 7 => {
495 column.total_compressed_size = i64::read_thrift(&mut *prot)?;
496 seen_mask |= COL_META_TOTAL_COMP_SZ;
497 }
498 9 => {
500 column.data_page_offset = i64::read_thrift(&mut *prot)?;
501 seen_mask |= COL_META_DATA_PAGE_OFFSET;
502 }
503 10 => {
504 column.index_page_offset = Some(i64::read_thrift(&mut *prot)?);
505 }
506 11 => {
507 column.dictionary_page_offset = Some(i64::read_thrift(&mut *prot)?);
508 }
509 12 if !skip_col_stats => {
510 column.statistics =
511 convert_stats(column_descr, Some(Statistics::read_thrift(&mut *prot)?))?;
512 }
513 13 if !skip_pes => {
514 if pes_mask {
515 let val = read_encoding_stats_as_mask(&mut *prot)?;
516 column.encoding_stats = Some(ParquetPageEncodingStats::Mask(val));
517 } else {
518 let val =
519 read_thrift_vec::<PageEncodingStats, ThriftSliceInputProtocol>(&mut *prot)?;
520 column.encoding_stats = Some(ParquetPageEncodingStats::Full(val));
521 }
522 }
523 14 => {
524 column.bloom_filter_offset = Some(i64::read_thrift(&mut *prot)?);
525 }
526 15 => {
527 column.bloom_filter_length = Some(i32::read_thrift(&mut *prot)?);
528 }
529 16 if !skip_size_stats => {
530 let val = SizeStatistics::read_thrift(&mut *prot)?;
531 column.unencoded_byte_array_data_bytes = val.unencoded_byte_array_data_bytes;
532 column.repetition_level_histogram =
533 val.repetition_level_histogram.map(LevelHistogram::from);
534 column.definition_level_histogram =
535 val.definition_level_histogram.map(LevelHistogram::from);
536 }
537 17 => {
538 let val = GeospatialStatistics::read_thrift(&mut *prot)?;
539 column.geo_statistics = Some(Box::new(convert_geo_stats(val)));
540 }
541 _ => {
542 prot.skip(field_ident.field_type)?;
543 }
544 };
545 last_field_id = field_ident.id;
546 }
547
548 Ok(seen_mask)
549}
550
551fn read_column_chunk<'a>(
554 prot: &mut ThriftSliceInputProtocol<'a>,
555 column_descr: &Arc<ColumnDescriptor>,
556 col_index: usize,
557 options: Option<&ParquetMetaDataOptions>,
558) -> Result<ColumnChunkMetaData> {
559 let mut col = ColumnChunkMetaDataBuilder::new(column_descr.clone()).build()?;
561
562 let mut has_file_offset = false;
564
565 let mut col_meta_mask = 0u16;
567
568 let mut last_field_id = 0i16;
580 loop {
581 let field_ident = prot.read_field_begin(last_field_id)?;
582 if field_ident.field_type == FieldType::Stop {
583 break;
584 }
585 match field_ident.id {
586 1 => {
587 col.file_path = Some(String::read_thrift(&mut *prot)?);
588 }
589 2 => {
590 col.file_offset = i64::read_thrift(&mut *prot)?;
591 has_file_offset = true;
592 }
593 3 => {
594 col_meta_mask = read_column_metadata(&mut *prot, &mut col, col_index, options)?;
595 }
596 4 => {
597 col.offset_index_offset = Some(i64::read_thrift(&mut *prot)?);
598 }
599 5 => {
600 col.offset_index_length = Some(i32::read_thrift(&mut *prot)?);
601 }
602 6 => {
603 col.column_index_offset = Some(i64::read_thrift(&mut *prot)?);
604 }
605 7 => {
606 col.column_index_length = Some(i32::read_thrift(&mut *prot)?);
607 }
608 #[cfg(feature = "encryption")]
609 8 => {
610 let val = ColumnCryptoMetaData::read_thrift(&mut *prot)?;
611 col.column_crypto_metadata = Some(Box::new(val));
612 }
613 #[cfg(feature = "encryption")]
614 9 => {
615 col.encrypted_column_metadata = Some(<&[u8]>::read_thrift(&mut *prot)?.to_vec());
616 }
617 _ => {
618 prot.skip(field_ident.field_type)?;
619 }
620 };
621 last_field_id = field_ident.id;
622 }
623
624 if !has_file_offset {
626 return Err(general_err!("Required field file_offset is missing"));
627 };
628
629 #[cfg(feature = "encryption")]
631 if col.encrypted_column_metadata.is_some() {
632 return Ok(col);
633 }
634
635 validate_column_metadata(col_meta_mask)?;
637
638 Ok(col)
639}
640
641fn read_row_group(
642 prot: &mut ThriftSliceInputProtocol,
643 schema_descr: &Arc<SchemaDescriptor>,
644 options: Option<&ParquetMetaDataOptions>,
645) -> Result<RowGroupMetaData> {
646 let mut row_group = RowGroupMetaDataBuilder::new(schema_descr.clone()).build_unchecked();
648
649 const RG_COLUMNS: u8 = 1 << 1;
651 const RG_TOT_BYTE_SIZE: u8 = 1 << 2;
652 const RG_NUM_ROWS: u8 = 1 << 3;
653 const RG_ALL_REQUIRED: u8 = RG_COLUMNS | RG_TOT_BYTE_SIZE | RG_NUM_ROWS;
654
655 let mut mask = 0u8;
656
657 let mut last_field_id = 0i16;
667 loop {
668 let field_ident = prot.read_field_begin(last_field_id)?;
669 if field_ident.field_type == FieldType::Stop {
670 break;
671 }
672 match field_ident.id {
673 1 => {
674 let list_ident = prot.read_list_begin()?;
675 validate_list_type(ElementType::Struct, &list_ident)?;
677 if schema_descr.num_columns() != list_ident.size as usize {
678 return Err(general_err!(
679 "Column count mismatch. Schema has {} columns while Row Group has {}",
680 schema_descr.num_columns(),
681 list_ident.size
682 ));
683 }
684 for i in 0..list_ident.size as usize {
685 let col = read_column_chunk(prot, &schema_descr.columns()[i], i, options)?;
686 row_group.columns.push(col);
687 }
688 mask |= RG_COLUMNS;
689 }
690 2 => {
691 row_group.total_byte_size = i64::read_thrift(&mut *prot)?;
692 mask |= RG_TOT_BYTE_SIZE;
693 }
694 3 => {
695 row_group.num_rows = i64::read_thrift(&mut *prot)?;
696 mask |= RG_NUM_ROWS;
697 }
698 4 => {
699 let val = read_thrift_vec::<SortingColumn, ThriftSliceInputProtocol>(&mut *prot)?;
700 row_group.sorting_columns = Some(val);
701 }
702 5 => {
703 row_group.file_offset = Some(i64::read_thrift(&mut *prot)?);
704 }
705 7 => {
707 row_group.ordinal = Some(i16::read_thrift(&mut *prot)? as i32);
708 }
709 _ => {
710 prot.skip(field_ident.field_type)?;
711 }
712 };
713 last_field_id = field_ident.id;
714 }
715
716 if mask != RG_ALL_REQUIRED {
717 if mask & RG_COLUMNS == 0 {
718 return Err(general_err!("Required field columns is missing"));
719 }
720 if mask & RG_TOT_BYTE_SIZE == 0 {
721 return Err(general_err!("Required field total_byte_size is missing"));
722 }
723 if mask & RG_NUM_ROWS == 0 {
724 return Err(general_err!("Required field num_rows is missing"));
725 }
726 }
727
728 Ok(row_group)
729}
730
731pub(crate) fn parquet_schema_from_bytes(buf: &[u8]) -> Result<SchemaDescriptor> {
734 let mut prot = ThriftSliceInputProtocol::new(buf);
735
736 let mut last_field_id = 0i16;
737 loop {
738 let field_ident = prot.read_field_begin(last_field_id)?;
739 if field_ident.field_type == FieldType::Stop {
740 break;
741 }
742 match field_ident.id {
743 2 => {
744 let val = read_thrift_vec::<SchemaElement, ThriftSliceInputProtocol>(&mut prot)?;
746 let val = parquet_schema_from_array(val)?;
747 return Ok(SchemaDescriptor::new(val));
748 }
749 _ => prot.skip(field_ident.field_type)?,
750 }
751 last_field_id = field_ident.id;
752 }
753 Err(general_err!("Input does not contain a schema"))
754}
755
756pub(crate) fn parquet_metadata_from_bytes(
759 buf: &[u8],
760 options: Option<&ParquetMetaDataOptions>,
761) -> Result<ParquetMetaData> {
762 let mut prot = ThriftSliceInputProtocol::new(buf);
763
764 let mut version: Option<i32> = None;
766 let mut num_rows: Option<i64> = None;
767 let mut row_groups: Option<Vec<RowGroupMetaData>> = None;
768 let mut key_value_metadata: Option<Vec<KeyValue>> = None;
769 let mut created_by: Option<&str> = None;
770 let mut column_orders: Option<Vec<ColumnOrder>> = None;
771 #[cfg(feature = "encryption")]
772 let mut encryption_algorithm: Option<EncryptionAlgorithm> = None;
773 #[cfg(feature = "encryption")]
774 let mut footer_signing_key_metadata: Option<&[u8]> = None;
775
776 let mut schema_descr: Option<Arc<SchemaDescriptor>> =
779 options.and_then(|options| options.schema().cloned());
780
781 let mut last_field_id = 0i16;
793 loop {
794 let field_ident = prot.read_field_begin(last_field_id)?;
795 if field_ident.field_type == FieldType::Stop {
796 break;
797 }
798 match field_ident.id {
799 1 => {
800 version = Some(i32::read_thrift(&mut prot)?);
801 }
802 2 => {
803 if schema_descr.is_some() {
805 prot.skip(field_ident.field_type)?;
806 } else {
807 let val =
809 read_thrift_vec::<SchemaElement, ThriftSliceInputProtocol>(&mut prot)?;
810 let val = parquet_schema_from_array(val)?;
811 schema_descr = Some(Arc::new(SchemaDescriptor::new(val)));
812 }
813 }
814 3 => {
815 num_rows = Some(i64::read_thrift(&mut prot)?);
816 }
817 4 => {
818 if schema_descr.is_none() {
819 return Err(general_err!("Required field schema is missing"));
820 }
821 let schema_descr = schema_descr.as_ref().unwrap();
822 let list_ident = prot.read_list_begin()?;
823 validate_list_type(ElementType::Struct, &list_ident)?;
825 let mut rg_vec = Vec::with_capacity(list_ident.size as usize);
826
827 for _ in 0..list_ident.size {
828 rg_vec.push(read_row_group(&mut prot, schema_descr, options)?);
829 }
830 ensure_row_group_ordinals(&mut rg_vec)?;
831 row_groups = Some(rg_vec);
832 }
833 5 => {
834 let val = read_thrift_vec::<KeyValue, ThriftSliceInputProtocol>(&mut prot)?;
835 key_value_metadata = Some(val);
836 }
837 6 => {
838 created_by = Some(<&str>::read_thrift(&mut prot)?);
839 }
840 7 => {
841 let val = read_thrift_vec::<ColumnOrder, ThriftSliceInputProtocol>(&mut prot)?;
842 column_orders = Some(val);
843 }
844 #[cfg(feature = "encryption")]
845 8 => {
846 let val = EncryptionAlgorithm::read_thrift(&mut prot)?;
847 encryption_algorithm = Some(val);
848 }
849 #[cfg(feature = "encryption")]
850 9 => {
851 footer_signing_key_metadata = Some(<&[u8]>::read_thrift(&mut prot)?);
852 }
853 _ => {
854 prot.skip(field_ident.field_type)?;
855 }
856 };
857 last_field_id = field_ident.id;
858 }
859 let Some(version) = version else {
860 return Err(general_err!("Required field version is missing"));
861 };
862 let Some(num_rows) = num_rows else {
863 return Err(general_err!("Required field num_rows is missing"));
864 };
865 let Some(row_groups) = row_groups else {
866 return Err(general_err!("Required field row_groups is missing"));
867 };
868
869 let created_by = created_by.map(|c| c.to_owned());
870
871 let schema_descr = schema_descr.unwrap();
873
874 if column_orders
876 .as_ref()
877 .is_some_and(|cos| cos.len() != schema_descr.num_columns())
878 {
879 return Err(general_err!("Column order length mismatch"));
880 }
881 let column_orders = column_orders.map(|mut cos| {
883 for (i, column) in schema_descr.columns().iter().enumerate() {
884 if let ColumnOrder::TYPE_DEFINED_ORDER(_) = cos[i] {
885 let sort_order = ColumnOrder::get_sort_order_for_type(
889 column.logical_type_ref(),
890 column.converted_type(),
891 column.physical_type(),
892 true,
893 );
894 cos[i] = ColumnOrder::TYPE_DEFINED_ORDER(sort_order);
895 }
896 }
897 cos
898 });
899
900 #[cfg(not(feature = "encryption"))]
901 let fmd = crate::file::metadata::FileMetaData::new(
902 version,
903 num_rows,
904 created_by,
905 key_value_metadata,
906 schema_descr,
907 column_orders,
908 );
909 #[cfg(feature = "encryption")]
910 let fmd = crate::file::metadata::FileMetaData::new(
911 version,
912 num_rows,
913 created_by,
914 key_value_metadata,
915 schema_descr,
916 column_orders,
917 )
918 .with_encryption_algorithm(encryption_algorithm)
919 .with_footer_signing_key_metadata(footer_signing_key_metadata.map(|v| v.to_vec()));
920
921 Ok(ParquetMetaData::new(fmd, row_groups))
922}
923
924fn ensure_row_group_ordinals(row_groups: &mut [RowGroupMetaData]) -> Result<()> {
941 if row_groups.iter().any(|rg| rg.ordinal.is_some()) {
945 return Ok(());
946 }
947 for (idx, rg) in row_groups.iter_mut().enumerate() {
948 let ordinal: i32 = idx
949 .try_into()
950 .map_err(|_| general_err!("Row group ordinal {} exceeds i32 max value", idx))?;
951 rg.ordinal = Some(ordinal);
952 }
953 Ok(())
954}
955
956thrift_struct!(
957 pub(crate) struct IndexPageHeader {}
958);
959
960thrift_struct!(
961pub(crate) struct DictionaryPageHeader {
962 1: required i32 num_values;
964
965 2: required Encoding encoding
967
968 3: optional bool is_sorted;
970}
971);
972
973thrift_struct!(
974pub(crate) struct PageStatistics {
982 1: optional binary max;
983 2: optional binary min;
984 3: optional i64 null_count;
985 4: optional i64 distinct_count;
986 5: optional binary max_value;
987 6: optional binary min_value;
988 7: optional bool is_max_value_exact;
989 8: optional bool is_min_value_exact;
990 9: optional i64 nan_count;
991}
992);
993
994thrift_struct!(
995pub(crate) struct DataPageHeader {
996 1: required i32 num_values
997 2: required Encoding encoding
998 3: required Encoding definition_level_encoding;
999 4: required Encoding repetition_level_encoding;
1000 5: optional PageStatistics statistics;
1001}
1002);
1003
1004impl DataPageHeader {
1005 fn read_thrift_without_stats<'a, R>(prot: &mut R) -> Result<Self>
1007 where
1008 R: ThriftCompactInputProtocol<'a>,
1009 {
1010 let mut num_values: Option<i32> = None;
1011 let mut encoding: Option<Encoding> = None;
1012 let mut definition_level_encoding: Option<Encoding> = None;
1013 let mut repetition_level_encoding: Option<Encoding> = None;
1014 let statistics: Option<PageStatistics> = None;
1015 let mut last_field_id = 0i16;
1016 loop {
1017 let field_ident = prot.read_field_begin(last_field_id)?;
1018 if field_ident.field_type == FieldType::Stop {
1019 break;
1020 }
1021 match field_ident.id {
1022 1 => {
1023 let val = i32::read_thrift(&mut *prot)?;
1024 num_values = Some(val);
1025 }
1026 2 => {
1027 let val = Encoding::read_thrift(&mut *prot)?;
1028 encoding = Some(val);
1029 }
1030 3 => {
1031 let val = Encoding::read_thrift(&mut *prot)?;
1032 definition_level_encoding = Some(val);
1033 }
1034 4 => {
1035 let val = Encoding::read_thrift(&mut *prot)?;
1036 repetition_level_encoding = Some(val);
1037 }
1038 _ => {
1039 prot.skip(field_ident.field_type)?;
1040 }
1041 };
1042 last_field_id = field_ident.id;
1043 }
1044 let Some(num_values) = num_values else {
1045 return Err(general_err!("Required field num_values is missing"));
1046 };
1047 let Some(encoding) = encoding else {
1048 return Err(general_err!("Required field encoding is missing"));
1049 };
1050 let Some(definition_level_encoding) = definition_level_encoding else {
1051 return Err(general_err!(
1052 "Required field definition_level_encoding is missing"
1053 ));
1054 };
1055 let Some(repetition_level_encoding) = repetition_level_encoding else {
1056 return Err(general_err!(
1057 "Required field repetition_level_encoding is missing"
1058 ));
1059 };
1060 Ok(Self {
1061 num_values,
1062 encoding,
1063 definition_level_encoding,
1064 repetition_level_encoding,
1065 statistics,
1066 })
1067 }
1068}
1069
1070thrift_struct!(
1071pub(crate) struct DataPageHeaderV2 {
1072 1: required i32 num_values
1073 2: required i32 num_nulls
1074 3: required i32 num_rows
1075 4: required Encoding encoding
1076 5: required i32 definition_levels_byte_length;
1077 6: required i32 repetition_levels_byte_length;
1078 7: optional bool is_compressed = true;
1079 8: optional PageStatistics statistics;
1080}
1081);
1082
1083impl DataPageHeaderV2 {
1084 fn read_thrift_without_stats<'a, R>(prot: &mut R) -> Result<Self>
1086 where
1087 R: ThriftCompactInputProtocol<'a>,
1088 {
1089 let mut num_values: Option<i32> = None;
1090 let mut num_nulls: Option<i32> = None;
1091 let mut num_rows: Option<i32> = None;
1092 let mut encoding: Option<Encoding> = None;
1093 let mut definition_levels_byte_length: Option<i32> = None;
1094 let mut repetition_levels_byte_length: Option<i32> = None;
1095 let mut is_compressed: Option<bool> = None;
1096 let statistics: Option<PageStatistics> = None;
1097 let mut last_field_id = 0i16;
1098 loop {
1099 let field_ident = prot.read_field_begin(last_field_id)?;
1100 if field_ident.field_type == FieldType::Stop {
1101 break;
1102 }
1103 match field_ident.id {
1104 1 => {
1105 let val = i32::read_thrift(&mut *prot)?;
1106 num_values = Some(val);
1107 }
1108 2 => {
1109 let val = i32::read_thrift(&mut *prot)?;
1110 num_nulls = Some(val);
1111 }
1112 3 => {
1113 let val = i32::read_thrift(&mut *prot)?;
1114 num_rows = Some(val);
1115 }
1116 4 => {
1117 let val = Encoding::read_thrift(&mut *prot)?;
1118 encoding = Some(val);
1119 }
1120 5 => {
1121 let val = i32::read_thrift(&mut *prot)?;
1122 definition_levels_byte_length = Some(val);
1123 }
1124 6 => {
1125 let val = i32::read_thrift(&mut *prot)?;
1126 repetition_levels_byte_length = Some(val);
1127 }
1128 7 => {
1129 is_compressed = Some(field_ident.bool_val()?);
1130 }
1131 _ => {
1132 prot.skip(field_ident.field_type)?;
1133 }
1134 };
1135 last_field_id = field_ident.id;
1136 }
1137 let Some(num_values) = num_values else {
1138 return Err(general_err!("Required field num_values is missing"));
1139 };
1140 let Some(num_nulls) = num_nulls else {
1141 return Err(general_err!("Required field num_nulls is missing"));
1142 };
1143 let Some(num_rows) = num_rows else {
1144 return Err(general_err!("Required field num_rows is missing"));
1145 };
1146 let Some(encoding) = encoding else {
1147 return Err(general_err!("Required field encoding is missing"));
1148 };
1149 let Some(definition_levels_byte_length) = definition_levels_byte_length else {
1150 return Err(general_err!(
1151 "Required field definition_levels_byte_length is missing"
1152 ));
1153 };
1154 let Some(repetition_levels_byte_length) = repetition_levels_byte_length else {
1155 return Err(general_err!(
1156 "Required field repetition_levels_byte_length is missing"
1157 ));
1158 };
1159 Ok(Self {
1160 num_values,
1161 num_nulls,
1162 num_rows,
1163 encoding,
1164 definition_levels_byte_length,
1165 repetition_levels_byte_length,
1166 is_compressed,
1167 statistics,
1168 })
1169 }
1170}
1171
1172thrift_struct!(
1173pub(crate) struct PageHeader {
1174 1: required PageType r#type
1176
1177 2: required i32 uncompressed_page_size
1179
1180 3: required i32 compressed_page_size
1182
1183 4: optional i32 crc
1185
1186 5: optional DataPageHeader data_page_header;
1188 6: optional IndexPageHeader index_page_header;
1189 7: optional DictionaryPageHeader dictionary_page_header;
1190 8: optional DataPageHeaderV2 data_page_header_v2;
1191}
1192);
1193
1194impl PageHeader {
1195 pub(crate) fn read_thrift_without_stats<'a, R>(prot: &mut R) -> Result<Self>
1199 where
1200 R: ThriftCompactInputProtocol<'a>,
1201 {
1202 let mut type_: Option<PageType> = None;
1203 let mut uncompressed_page_size: Option<i32> = None;
1204 let mut compressed_page_size: Option<i32> = None;
1205 let mut crc: Option<i32> = None;
1206 let mut data_page_header: Option<DataPageHeader> = None;
1207 let mut index_page_header: Option<IndexPageHeader> = None;
1208 let mut dictionary_page_header: Option<DictionaryPageHeader> = None;
1209 let mut data_page_header_v2: Option<DataPageHeaderV2> = None;
1210 let mut last_field_id = 0i16;
1211 loop {
1212 let field_ident = prot.read_field_begin(last_field_id)?;
1213 if field_ident.field_type == FieldType::Stop {
1214 break;
1215 }
1216 match field_ident.id {
1217 1 => {
1218 let val = PageType::read_thrift(&mut *prot)?;
1219 type_ = Some(val);
1220 }
1221 2 => {
1222 let val = i32::read_thrift(&mut *prot)?;
1223 uncompressed_page_size = Some(val);
1224 }
1225 3 => {
1226 let val = i32::read_thrift(&mut *prot)?;
1227 compressed_page_size = Some(val);
1228 }
1229 4 => {
1230 let val = i32::read_thrift(&mut *prot)?;
1231 crc = Some(val);
1232 }
1233 5 => {
1234 let val = DataPageHeader::read_thrift_without_stats(&mut *prot)?;
1235 data_page_header = Some(val);
1236 }
1237 6 => {
1238 let val = IndexPageHeader::read_thrift(&mut *prot)?;
1239 index_page_header = Some(val);
1240 }
1241 7 => {
1242 let val = DictionaryPageHeader::read_thrift(&mut *prot)?;
1243 dictionary_page_header = Some(val);
1244 }
1245 8 => {
1246 let val = DataPageHeaderV2::read_thrift_without_stats(&mut *prot)?;
1247 data_page_header_v2 = Some(val);
1248 }
1249 _ => {
1250 prot.skip(field_ident.field_type)?;
1251 }
1252 };
1253 last_field_id = field_ident.id;
1254 }
1255 let Some(type_) = type_ else {
1256 return Err(general_err!("Required field type_ is missing"));
1257 };
1258 let Some(uncompressed_page_size) = uncompressed_page_size else {
1259 return Err(general_err!(
1260 "Required field uncompressed_page_size is missing"
1261 ));
1262 };
1263 let Some(compressed_page_size) = compressed_page_size else {
1264 return Err(general_err!(
1265 "Required field compressed_page_size is missing"
1266 ));
1267 };
1268 Ok(Self {
1269 r#type: type_,
1270 uncompressed_page_size,
1271 compressed_page_size,
1272 crc,
1273 data_page_header,
1274 index_page_header,
1275 dictionary_page_header,
1276 data_page_header_v2,
1277 })
1278 }
1279}
1280
1281#[cfg(feature = "encryption")]
1285fn should_write_column_stats(column_chunk: &ColumnChunkMetaData) -> bool {
1286 column_chunk.encrypted_column_metadata.is_none()
1290}
1291
1292#[cfg(not(feature = "encryption"))]
1293fn should_write_column_stats(_column_chunk: &ColumnChunkMetaData) -> bool {
1294 true
1295}
1296
1297pub(super) fn serialize_column_meta_data<W: Write>(
1318 column_chunk: &ColumnChunkMetaData,
1319 w: &mut ThriftCompactOutputProtocol<W>,
1320) -> Result<()> {
1321 use crate::file::statistics::page_stats_to_thrift;
1322
1323 column_chunk.column_type().write_thrift_field(w, 1, 0)?;
1324 column_chunk
1325 .encodings()
1326 .collect::<Vec<_>>()
1327 .write_thrift_field(w, 2, 1)?;
1328 if w.write_path_in_schema() {
1329 let path = column_chunk.column_descr.path().parts();
1330 let path: Vec<&str> = path.iter().map(|v| v.as_str()).collect();
1331 path.write_thrift_field(w, 3, 2)?;
1332 column_chunk.compression.write_thrift_field(w, 4, 3)?;
1333 } else {
1334 column_chunk.compression.write_thrift_field(w, 4, 2)?;
1335 }
1336
1337 column_chunk.num_values.write_thrift_field(w, 5, 4)?;
1338 column_chunk
1339 .total_uncompressed_size
1340 .write_thrift_field(w, 6, 5)?;
1341 column_chunk
1342 .total_compressed_size
1343 .write_thrift_field(w, 7, 6)?;
1344 let mut last_field_id = column_chunk.data_page_offset.write_thrift_field(w, 9, 7)?;
1346 if let Some(index_page_offset) = column_chunk.index_page_offset {
1347 last_field_id = index_page_offset.write_thrift_field(w, 10, last_field_id)?;
1348 }
1349 if let Some(dictionary_page_offset) = column_chunk.dictionary_page_offset {
1350 last_field_id = dictionary_page_offset.write_thrift_field(w, 11, last_field_id)?;
1351 }
1352
1353 if should_write_column_stats(column_chunk) {
1354 let stats = page_stats_to_thrift(column_chunk.statistics());
1356 if let Some(stats) = stats {
1357 last_field_id = stats.write_thrift_field(w, 12, last_field_id)?;
1358 }
1359 if let Some(page_encoding_stats) = column_chunk.page_encoding_stats() {
1360 last_field_id = page_encoding_stats.write_thrift_field(w, 13, last_field_id)?;
1361 }
1362 if let Some(bloom_filter_offset) = column_chunk.bloom_filter_offset {
1363 last_field_id = bloom_filter_offset.write_thrift_field(w, 14, last_field_id)?;
1364 }
1365 if let Some(bloom_filter_length) = column_chunk.bloom_filter_length {
1366 last_field_id = bloom_filter_length.write_thrift_field(w, 15, last_field_id)?;
1367 }
1368
1369 let size_stats = if column_chunk.unencoded_byte_array_data_bytes.is_some()
1371 || column_chunk.repetition_level_histogram.is_some()
1372 || column_chunk.definition_level_histogram.is_some()
1373 {
1374 let repetition_level_histogram = column_chunk
1375 .repetition_level_histogram()
1376 .map(|hist| hist.clone().into_inner());
1377
1378 let definition_level_histogram = column_chunk
1379 .definition_level_histogram()
1380 .map(|hist| hist.clone().into_inner());
1381
1382 Some(SizeStatistics {
1383 unencoded_byte_array_data_bytes: column_chunk.unencoded_byte_array_data_bytes,
1384 repetition_level_histogram,
1385 definition_level_histogram,
1386 })
1387 } else {
1388 None
1389 };
1390 if let Some(size_stats) = size_stats {
1391 last_field_id = size_stats.write_thrift_field(w, 16, last_field_id)?;
1392 }
1393
1394 if let Some(geo_stats) = column_chunk.geo_statistics() {
1395 geo_stats.write_thrift_field(w, 17, last_field_id)?;
1396 }
1397 }
1398
1399 w.write_struct_end()
1400}
1401
1402pub(super) struct FileMeta<'a> {
1404 pub(super) file_metadata: &'a crate::file::metadata::FileMetaData,
1405 pub(super) row_groups: &'a Vec<RowGroupMetaData>,
1406 pub(super) write_path_in_schema: bool,
1408}
1409
1410impl<'a> WriteThrift for FileMeta<'a> {
1422 const ELEMENT_TYPE: ElementType = ElementType::Struct;
1423
1424 #[allow(unused_assignments)]
1426 fn write_thrift<W: Write>(&self, writer: &mut ThriftCompactOutputProtocol<W>) -> Result<()> {
1427 writer.set_write_path_in_schema(self.write_path_in_schema);
1428 writer.set_write_row_group_ordinal(i16::try_from(self.row_groups.len()).is_ok());
1430
1431 self.file_metadata
1432 .version
1433 .write_thrift_field(writer, 1, 0)?;
1434
1435 let root = self.file_metadata.schema_descr().root_schema_ptr();
1438 let schema_len = num_nodes(&root)?;
1439 writer.write_field_begin(FieldType::List, 2, 1)?;
1440 writer.write_list_begin(ElementType::Struct, schema_len)?;
1441 write_schema(&root, writer)?;
1443
1444 self.file_metadata
1445 .num_rows
1446 .write_thrift_field(writer, 3, 2)?;
1447
1448 let mut last_field_id = self.row_groups.write_thrift_field(writer, 4, 3)?;
1450
1451 if let Some(kv_metadata) = self.file_metadata.key_value_metadata() {
1452 last_field_id = kv_metadata.write_thrift_field(writer, 5, last_field_id)?;
1453 }
1454 if let Some(created_by) = self.file_metadata.created_by() {
1455 last_field_id = created_by.write_thrift_field(writer, 6, last_field_id)?;
1456 }
1457 if let Some(column_orders) = self.file_metadata.column_orders() {
1458 last_field_id = column_orders.write_thrift_field(writer, 7, last_field_id)?;
1459 }
1460 #[cfg(feature = "encryption")]
1461 if let Some(algo) = self.file_metadata.encryption_algorithm.as_ref() {
1462 last_field_id = algo.write_thrift_field(writer, 8, last_field_id)?;
1463 }
1464 #[cfg(feature = "encryption")]
1465 if let Some(key) = self.file_metadata.footer_signing_key_metadata.as_ref() {
1466 key.as_slice()
1467 .write_thrift_field(writer, 9, last_field_id)?;
1468 }
1469
1470 writer.write_struct_end()
1471 }
1472}
1473
1474fn write_schema<W: Write>(
1475 schema: &TypePtr,
1476 writer: &mut ThriftCompactOutputProtocol<W>,
1477) -> Result<()> {
1478 if !schema.is_group() {
1479 return Err(general_err!("Root schema must be Group type"));
1480 }
1481 write_schema_helper(schema, writer)
1482}
1483
1484fn write_schema_helper<W: Write>(
1485 node: &TypePtr,
1486 writer: &mut ThriftCompactOutputProtocol<W>,
1487) -> Result<()> {
1488 match node.as_ref() {
1489 crate::schema::types::Type::PrimitiveType {
1490 basic_info,
1491 physical_type,
1492 type_length,
1493 scale,
1494 precision,
1495 } => {
1496 let element = SchemaElement {
1497 r#type: Some(*physical_type),
1498 type_length: if *type_length >= 0 {
1499 Some(*type_length)
1500 } else {
1501 None
1502 },
1503 repetition_type: Some(basic_info.repetition()),
1504 name: basic_info.name(),
1505 num_children: None,
1506 converted_type: match basic_info.converted_type() {
1507 ConvertedType::NONE => None,
1508 other => Some(other),
1509 },
1510 scale: if *scale >= 0 { Some(*scale) } else { None },
1511 precision: if *precision >= 0 {
1512 Some(*precision)
1513 } else {
1514 None
1515 },
1516 field_id: if basic_info.has_id() {
1517 Some(basic_info.id())
1518 } else {
1519 None
1520 },
1521 logical_type: basic_info.logical_type_ref().cloned(),
1522 };
1523 element.write_thrift(writer)
1524 }
1525 crate::schema::types::Type::GroupType { basic_info, fields } => {
1526 let repetition = if basic_info.has_repetition() {
1527 Some(basic_info.repetition())
1528 } else {
1529 None
1530 };
1531
1532 let element = SchemaElement {
1533 r#type: None,
1534 type_length: None,
1535 repetition_type: repetition,
1536 name: basic_info.name(),
1537 num_children: Some(fields.len().try_into()?),
1538 converted_type: match basic_info.converted_type() {
1539 ConvertedType::NONE => None,
1540 other => Some(other),
1541 },
1542 scale: None,
1543 precision: None,
1544 field_id: if basic_info.has_id() {
1545 Some(basic_info.id())
1546 } else {
1547 None
1548 },
1549 logical_type: basic_info.logical_type_ref().cloned(),
1550 };
1551
1552 element.write_thrift(writer)?;
1553
1554 for field in fields {
1556 write_schema_helper(field, writer)?;
1557 }
1558 Ok(())
1559 }
1560 }
1561}
1562
1563impl WriteThrift for RowGroupMetaData {
1573 const ELEMENT_TYPE: ElementType = ElementType::Struct;
1574
1575 fn write_thrift<W: Write>(&self, writer: &mut ThriftCompactOutputProtocol<W>) -> Result<()> {
1576 self.columns.write_thrift_field(writer, 1, 0)?;
1578 self.total_byte_size.write_thrift_field(writer, 2, 1)?;
1579 let mut last_field_id = self.num_rows.write_thrift_field(writer, 3, 2)?;
1580 if let Some(sorting_columns) = self.sorting_columns() {
1581 last_field_id = sorting_columns.write_thrift_field(writer, 4, last_field_id)?;
1582 }
1583 if let Some(file_offset) = self.file_offset() {
1584 last_field_id = file_offset.write_thrift_field(writer, 5, last_field_id)?;
1585 }
1586 last_field_id = self
1588 .compressed_size()
1589 .write_thrift_field(writer, 6, last_field_id)?;
1590
1591 if writer.write_row_group_ordinal()
1593 && let Some(ordinal) = self.ordinal()
1594 && let Ok(ordinal) = i16::try_from(ordinal)
1595 {
1596 ordinal.write_thrift_field(writer, 7, last_field_id)?;
1597 }
1598 writer.write_struct_end()
1599 }
1600}
1601
1602impl WriteThrift for ColumnChunkMetaData {
1614 const ELEMENT_TYPE: ElementType = ElementType::Struct;
1615
1616 #[allow(unused_assignments)]
1617 fn write_thrift<W: Write>(&self, writer: &mut ThriftCompactOutputProtocol<W>) -> Result<()> {
1618 let mut last_field_id = 0i16;
1619 if let Some(file_path) = self.file_path() {
1620 last_field_id = file_path.write_thrift_field(writer, 1, last_field_id)?;
1621 }
1622 last_field_id = self
1623 .file_offset()
1624 .write_thrift_field(writer, 2, last_field_id)?;
1625
1626 #[cfg(feature = "encryption")]
1627 let write_meta_data =
1628 self.encrypted_column_metadata.is_none() || self.plaintext_footer_mode;
1629 #[cfg(not(feature = "encryption"))]
1630 let write_meta_data = true;
1631
1632 if write_meta_data {
1638 writer.write_field_begin(FieldType::Struct, 3, last_field_id)?;
1639 serialize_column_meta_data(self, writer)?;
1640 last_field_id = 3;
1641 }
1642
1643 if let Some(offset_idx_off) = self.offset_index_offset() {
1644 last_field_id = offset_idx_off.write_thrift_field(writer, 4, last_field_id)?;
1645 }
1646 if let Some(offset_idx_len) = self.offset_index_length() {
1647 last_field_id = offset_idx_len.write_thrift_field(writer, 5, last_field_id)?;
1648 }
1649 if let Some(column_idx_off) = self.column_index_offset() {
1650 last_field_id = column_idx_off.write_thrift_field(writer, 6, last_field_id)?;
1651 }
1652 if let Some(column_idx_len) = self.column_index_length() {
1653 last_field_id = column_idx_len.write_thrift_field(writer, 7, last_field_id)?;
1654 }
1655 #[cfg(feature = "encryption")]
1656 {
1657 if let Some(crypto_metadata) = self.crypto_metadata() {
1658 last_field_id = crypto_metadata.write_thrift_field(writer, 8, last_field_id)?;
1659 }
1660 if let Some(encrypted_meta) = self.encrypted_column_metadata.as_ref() {
1661 encrypted_meta
1662 .as_slice()
1663 .write_thrift_field(writer, 9, last_field_id)?;
1664 }
1665 }
1666
1667 writer.write_struct_end()
1668 }
1669}
1670
1671impl WriteThrift for crate::geospatial::statistics::GeospatialStatistics {
1676 const ELEMENT_TYPE: ElementType = ElementType::Struct;
1677
1678 fn write_thrift<W: Write>(&self, writer: &mut ThriftCompactOutputProtocol<W>) -> Result<()> {
1679 let mut last_field_id = 0i16;
1680 if let Some(bbox) = self.bounding_box() {
1681 last_field_id = bbox.write_thrift_field(writer, 1, last_field_id)?;
1682 }
1683 if let Some(geo_types) = self.geospatial_types() {
1684 geo_types.write_thrift_field(writer, 2, last_field_id)?;
1685 }
1686
1687 writer.write_struct_end()
1688 }
1689}
1690
1691use crate::geospatial::statistics::GeospatialStatistics as RustGeospatialStatistics;
1693write_thrift_field!(RustGeospatialStatistics, FieldType::Struct);
1694
1695impl WriteThrift for crate::geospatial::bounding_box::BoundingBox {
1706 const ELEMENT_TYPE: ElementType = ElementType::Struct;
1707
1708 fn write_thrift<W: Write>(&self, writer: &mut ThriftCompactOutputProtocol<W>) -> Result<()> {
1709 self.get_xmin().write_thrift_field(writer, 1, 0)?;
1710 self.get_xmax().write_thrift_field(writer, 2, 1)?;
1711 self.get_ymin().write_thrift_field(writer, 3, 2)?;
1712 let mut last_field_id = self.get_ymax().write_thrift_field(writer, 4, 3)?;
1713
1714 if let Some(zmin) = self.get_zmin() {
1715 last_field_id = zmin.write_thrift_field(writer, 5, last_field_id)?;
1716 }
1717 if let Some(zmax) = self.get_zmax() {
1718 last_field_id = zmax.write_thrift_field(writer, 6, last_field_id)?;
1719 }
1720 if let Some(mmin) = self.get_mmin() {
1721 last_field_id = mmin.write_thrift_field(writer, 7, last_field_id)?;
1722 }
1723 if let Some(mmax) = self.get_mmax() {
1724 mmax.write_thrift_field(writer, 8, last_field_id)?;
1725 }
1726
1727 writer.write_struct_end()
1728 }
1729}
1730
1731use crate::geospatial::bounding_box::BoundingBox as RustBoundingBox;
1733write_thrift_field!(RustBoundingBox, FieldType::Struct);
1734
1735#[cfg(test)]
1736pub(crate) mod tests {
1737 use crate::basic::{Encoding, PageType, Type as PhysicalType};
1738 use crate::errors::Result;
1739 use crate::file::metadata::thrift::{
1740 BoundingBox, DataPageHeaderV2, DictionaryPageHeader, PageHeader, SchemaElement,
1741 write_schema,
1742 };
1743 use crate::file::metadata::{ColumnChunkMetaData, ParquetMetaDataOptions, RowGroupMetaData};
1744 use crate::parquet_thrift::tests::test_roundtrip;
1745 use crate::parquet_thrift::{
1746 ElementType, ThriftCompactOutputProtocol, ThriftSliceInputProtocol, WriteThrift,
1747 read_thrift_vec,
1748 };
1749 use crate::schema::types::{
1750 ColumnDescriptor, ColumnPath, SchemaDescriptor, TypePtr, num_nodes,
1751 parquet_schema_from_array,
1752 };
1753 use std::sync::Arc;
1754
1755 pub(crate) fn read_row_group(
1757 buf: &mut [u8],
1758 schema_descr: Arc<SchemaDescriptor>,
1759 ) -> Result<RowGroupMetaData> {
1760 let mut reader = ThriftSliceInputProtocol::new(buf);
1761 crate::file::metadata::thrift::read_row_group(&mut reader, &schema_descr, None)
1762 }
1763
1764 pub(crate) fn read_column_chunk(
1765 buf: &mut [u8],
1766 column_descr: Arc<ColumnDescriptor>,
1767 ) -> Result<ColumnChunkMetaData> {
1768 read_column_chunk_with_options(buf, column_descr, None)
1769 }
1770
1771 pub(crate) fn read_column_chunk_with_options(
1772 buf: &mut [u8],
1773 column_descr: Arc<ColumnDescriptor>,
1774 options: Option<&ParquetMetaDataOptions>,
1775 ) -> Result<ColumnChunkMetaData> {
1776 let mut reader = ThriftSliceInputProtocol::new(buf);
1777 crate::file::metadata::thrift::read_column_chunk(&mut reader, &column_descr, 0, options)
1778 }
1779
1780 pub(crate) fn roundtrip_schema(schema: TypePtr) -> Result<TypePtr> {
1781 let num_nodes = num_nodes(&schema)?;
1782 let mut buf = Vec::new();
1783 let mut writer = ThriftCompactOutputProtocol::new(&mut buf);
1784
1785 writer.write_list_begin(ElementType::Struct, num_nodes)?;
1787
1788 write_schema(&schema, &mut writer)?;
1790
1791 let mut prot = ThriftSliceInputProtocol::new(&buf);
1792 let se: Vec<SchemaElement> = read_thrift_vec(&mut prot)?;
1793 parquet_schema_from_array(se)
1794 }
1795
1796 pub(crate) fn schema_to_buf(schema: &TypePtr) -> Result<Vec<u8>> {
1797 let num_nodes = num_nodes(schema)?;
1798 let mut buf = Vec::new();
1799 let mut writer = ThriftCompactOutputProtocol::new(&mut buf);
1800
1801 writer.write_list_begin(ElementType::Struct, num_nodes)?;
1803
1804 write_schema(schema, &mut writer)?;
1806 Ok(buf)
1807 }
1808
1809 pub(crate) fn buf_to_schema_list<'a>(buf: &'a mut Vec<u8>) -> Result<Vec<SchemaElement<'a>>> {
1810 let mut prot = ThriftSliceInputProtocol::new(buf.as_mut_slice());
1811 read_thrift_vec(&mut prot)
1812 }
1813
1814 fn thrift_bytes<T: WriteThrift>(value: &T) -> Vec<u8> {
1815 let mut buf = Vec::new();
1816 let mut writer = ThriftCompactOutputProtocol::new(&mut buf);
1817 value.write_thrift(&mut writer).unwrap();
1818 buf
1819 }
1820
1821 fn change_false_bool_field_to_i32(buf: &mut [u8]) {
1822 let pos = buf
1823 .iter()
1824 .rposition(|byte| *byte == 0x12)
1825 .expect("expected BOOL_FALSE field header byte");
1826 buf[pos] = 0x15;
1827 }
1828
1829 fn assert_malformed_bool_error(err: crate::errors::ParquetError) {
1830 let msg = err.to_string();
1831 assert!(
1832 msg.contains("Unexpected struct field type"),
1833 "unexpected error message: {msg}"
1834 );
1835 }
1836
1837 #[test]
1838 fn test_bounding_box_roundtrip() {
1839 test_roundtrip(BoundingBox {
1840 xmin: 0.1.into(),
1841 xmax: 10.3.into(),
1842 ymin: 0.001.into(),
1843 ymax: 128.5.into(),
1844 zmin: None,
1845 zmax: None,
1846 mmin: None,
1847 mmax: None,
1848 });
1849
1850 test_roundtrip(BoundingBox {
1851 xmin: 0.1.into(),
1852 xmax: 10.3.into(),
1853 ymin: 0.001.into(),
1854 ymax: 128.5.into(),
1855 zmin: Some(11.0.into()),
1856 zmax: Some(1300.0.into()),
1857 mmin: None,
1858 mmax: None,
1859 });
1860
1861 test_roundtrip(BoundingBox {
1862 xmin: 0.1.into(),
1863 xmax: 10.3.into(),
1864 ymin: 0.001.into(),
1865 ymax: 128.5.into(),
1866 zmin: Some(11.0.into()),
1867 zmax: Some(1300.0.into()),
1868 mmin: Some(3.7.into()),
1869 mmax: Some(42.0.into()),
1870 });
1871 }
1872
1873 #[test]
1874 fn test_convert_stats_preserves_missing_null_count() {
1875 let primitive =
1876 crate::schema::types::Type::primitive_type_builder("col", PhysicalType::INT32)
1877 .build()
1878 .unwrap();
1879 let column_descr = Arc::new(ColumnDescriptor::new(
1880 Arc::new(primitive),
1881 0,
1882 0,
1883 ColumnPath::new(vec![]),
1884 ));
1885
1886 let none_null_count = super::Statistics {
1887 max: None,
1888 min: None,
1889 null_count: None,
1890 distinct_count: None,
1891 max_value: None,
1892 min_value: None,
1893 is_max_value_exact: None,
1894 is_min_value_exact: None,
1895 nan_count: None,
1896 };
1897 let decoded_none = super::convert_stats(&column_descr, Some(none_null_count))
1898 .unwrap()
1899 .unwrap();
1900 assert_eq!(decoded_none.null_count_opt(), None);
1901
1902 let zero_null_count = super::Statistics {
1903 max: None,
1904 min: None,
1905 null_count: Some(0),
1906 distinct_count: None,
1907 max_value: None,
1908 min_value: None,
1909 is_max_value_exact: None,
1910 is_min_value_exact: None,
1911 nan_count: None,
1912 };
1913 let decoded_zero = super::convert_stats(&column_descr, Some(zero_null_count))
1914 .unwrap()
1915 .unwrap();
1916 assert_eq!(decoded_zero.null_count_opt(), Some(0));
1917 }
1918
1919 #[test]
1920 fn test_convert_stats_returns_error_for_overlong_int96_statistics() {
1921 let primitive =
1922 crate::schema::types::Type::primitive_type_builder("col", PhysicalType::INT96)
1923 .build()
1924 .unwrap();
1925 let column_descr = Arc::new(ColumnDescriptor::new(
1926 Arc::new(primitive),
1927 0,
1928 0,
1929 ColumnPath::new(vec![]),
1930 ));
1931 let invalid = (0..13).collect::<Vec<_>>();
1932
1933 let make_stats = |min, max| super::Statistics {
1934 max,
1935 min,
1936 null_count: Some(0),
1937 distinct_count: None,
1938 max_value: None,
1939 min_value: None,
1940 is_max_value_exact: None,
1941 is_min_value_exact: None,
1942 nan_count: None,
1943 };
1944
1945 let err = super::convert_stats(&column_descr, Some(make_stats(Some(&invalid), None)))
1946 .unwrap_err();
1947 assert_eq!(
1948 err.to_string(),
1949 "Parquet error: Incorrect Int96 min statistics"
1950 );
1951
1952 let err = super::convert_stats(&column_descr, Some(make_stats(None, Some(&invalid))))
1953 .unwrap_err();
1954 assert_eq!(
1955 err.to_string(),
1956 "Parquet error: Incorrect Int96 max statistics"
1957 );
1958 }
1959
1960 #[test]
1961 fn malformed_bool_field_returns_error_not_panic() {
1962 let page_header = PageHeader {
1963 r#type: PageType::DICTIONARY_PAGE,
1964 uncompressed_page_size: 1,
1965 compressed_page_size: 1,
1966 crc: None,
1967 data_page_header: None,
1968 index_page_header: None,
1969 dictionary_page_header: Some(DictionaryPageHeader {
1970 num_values: 1,
1971 encoding: Encoding::PLAIN,
1972 is_sorted: Some(false),
1973 }),
1974 data_page_header_v2: None,
1975 };
1976
1977 let mut buf = thrift_bytes(&page_header);
1978 change_false_bool_field_to_i32(&mut buf);
1979
1980 let mut prot = ThriftSliceInputProtocol::new(&buf);
1981 let err = PageHeader::read_thrift_without_stats(&mut prot)
1982 .expect_err("malformed bool field should return an error");
1983 assert_malformed_bool_error(err);
1984 }
1985
1986 #[test]
1987 fn malformed_data_page_v2_bool_field_returns_error_not_panic() {
1988 let data_page_header_v2 = DataPageHeaderV2 {
1989 num_values: 1,
1990 num_nulls: 0,
1991 num_rows: 1,
1992 encoding: Encoding::PLAIN,
1993 definition_levels_byte_length: 0,
1994 repetition_levels_byte_length: 0,
1995 is_compressed: Some(false),
1996 statistics: None,
1997 };
1998
1999 let mut buf = thrift_bytes(&data_page_header_v2);
2000 change_false_bool_field_to_i32(&mut buf);
2001
2002 let mut prot = ThriftSliceInputProtocol::new(&buf);
2003 let err = DataPageHeaderV2::read_thrift_without_stats(&mut prot)
2004 .expect_err("malformed bool field should return an error");
2005 assert_malformed_bool_error(err);
2006 }
2007
2008 fn roundtrip_rg_ordinals(ordinals: &[Option<i32>]) -> Vec<Option<i32>> {
2012 use crate::file::metadata::ParquetMetaDataWriter;
2013 use crate::file::metadata::{FileMetaData, ParquetMetaData, ParquetMetaDataReader};
2014 use crate::schema::types::Type as SchemaType;
2015
2016 let field = SchemaType::primitive_type_builder("c", PhysicalType::INT32)
2017 .build()
2018 .unwrap();
2019 let schema = SchemaType::group_type_builder("schema")
2020 .with_fields(vec![Arc::new(field)])
2021 .build()
2022 .unwrap();
2023 let schema_descr = Arc::new(SchemaDescriptor::new(Arc::new(schema)));
2024
2025 let row_groups = ordinals
2026 .iter()
2027 .map(|ordinal| {
2028 let columns = schema_descr
2029 .columns()
2030 .iter()
2031 .map(|col| ColumnChunkMetaData::builder(col.clone()).build().unwrap())
2032 .collect();
2033 let mut builder =
2034 crate::file::metadata::RowGroupMetaData::builder(schema_descr.clone())
2035 .set_num_rows(10)
2036 .set_total_byte_size(100)
2037 .set_column_metadata(columns);
2038 if let Some(ordinal) = ordinal {
2039 builder = builder.set_ordinal(*ordinal);
2040 }
2041 builder.build().unwrap()
2042 })
2043 .collect();
2044
2045 let file_metadata = FileMetaData::new(
2046 1,
2047 10 * ordinals.len() as i64,
2048 None,
2049 None,
2050 schema_descr,
2051 None,
2052 );
2053 let metadata = ParquetMetaData::new(file_metadata, row_groups);
2054
2055 let mut buffer = Vec::new();
2056 ParquetMetaDataWriter::new(&mut buffer, &metadata)
2057 .finish()
2058 .unwrap();
2059 let decoded = ParquetMetaDataReader::decode_metadata(&buffer[..buffer.len() - 8]).unwrap();
2061 decoded.row_groups().iter().map(|rg| rg.ordinal()).collect()
2062 }
2063
2064 #[test]
2067 fn ordinals_all_present_are_honored() {
2068 assert_eq!(
2069 roundtrip_rg_ordinals(&[Some(5), Some(1), Some(3)]),
2070 vec![Some(5), Some(1), Some(3)],
2071 );
2072 }
2073
2074 #[test]
2077 fn ordinals_none_present_are_sequentially_filled() {
2078 assert_eq!(
2079 roundtrip_rg_ordinals(&[None, None, None]),
2080 vec![Some(0), Some(1), Some(2)],
2081 );
2082 }
2083
2084 #[test]
2089 fn ordinals_mixed_decode_succeeds_untouched() {
2090 assert_eq!(
2091 roundtrip_rg_ordinals(&[Some(0), None, Some(2)]),
2092 vec![Some(0), None, Some(2)],
2093 );
2094 assert_eq!(
2096 roundtrip_rg_ordinals(&[None, Some(1), Some(2)]),
2097 vec![None, Some(1), Some(2)],
2098 );
2099 }
2100}