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, Compression, ConvertedType, Encoding, EncodingMask, LogicalType, PageType,
39 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, RowGroupMetaData, RowGroupMetaDataBuilder,
47 SortingColumn,
48 },
49 statistics::ValueStatistics,
50 },
51 parquet_thrift::{
52 ElementType, FieldType, ReadThrift, ThriftCompactInputProtocol,
53 ThriftCompactOutputProtocol, ThriftSliceInputProtocol, WriteThrift, WriteThriftField,
54 read_thrift_vec,
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}
117);
118
119thrift_struct!(
120struct BoundingBox {
121 1: required double xmin;
122 2: required double xmax;
123 3: required double ymin;
124 4: required double ymax;
125 5: optional double zmin;
126 6: optional double zmax;
127 7: optional double mmin;
128 8: optional double mmax;
129}
130);
131
132thrift_struct!(
133struct GeospatialStatistics {
134 1: optional BoundingBox bbox;
135 2: optional list<i32> geospatial_types;
136}
137);
138
139thrift_struct!(
140struct SizeStatistics {
141 1: optional i64 unencoded_byte_array_data_bytes;
142 2: optional list<i64> repetition_level_histogram;
143 3: optional list<i64> definition_level_histogram;
144}
145);
146
147fn convert_geo_stats(
148 stats: Option<GeospatialStatistics>,
149) -> Option<Box<crate::geospatial::statistics::GeospatialStatistics>> {
150 stats.map(|st| {
151 let bbox = convert_bounding_box(st.bbox);
152 let geospatial_types: Option<Vec<i32>> = st.geospatial_types.filter(|v| !v.is_empty());
153 Box::new(crate::geospatial::statistics::GeospatialStatistics::new(
154 bbox,
155 geospatial_types,
156 ))
157 })
158}
159
160fn convert_bounding_box(
161 bbox: Option<BoundingBox>,
162) -> Option<crate::geospatial::bounding_box::BoundingBox> {
163 bbox.map(|bb| {
164 let mut newbb = crate::geospatial::bounding_box::BoundingBox::new(
165 bb.xmin.into(),
166 bb.xmax.into(),
167 bb.ymin.into(),
168 bb.ymax.into(),
169 );
170
171 newbb = match (bb.zmin, bb.zmax) {
172 (Some(zmin), Some(zmax)) => newbb.with_zrange(zmin.into(), zmax.into()),
173 _ => newbb,
175 };
176
177 newbb = match (bb.mmin, bb.mmax) {
178 (Some(mmin), Some(mmax)) => newbb.with_mrange(mmin.into(), mmax.into()),
179 _ => newbb,
181 };
182
183 newbb
184 })
185}
186
187fn convert_stats(
189 column_descr: &Arc<ColumnDescriptor>,
190 thrift_stats: Option<Statistics>,
191) -> Result<Option<crate::file::statistics::Statistics>> {
192 use crate::file::statistics::Statistics as FStatistics;
193 Ok(match thrift_stats {
194 Some(stats) => {
195 let null_count = stats.null_count.unwrap_or(0);
199
200 if null_count < 0 {
201 return Err(general_err!(
202 "Statistics null count is negative {}",
203 null_count
204 ));
205 }
206
207 let null_count = Some(null_count as u64);
209 let distinct_count = stats.distinct_count.map(|value| value as u64);
211 let old_format = stats.min_value.is_none() && stats.max_value.is_none();
213 let min = if old_format {
215 stats.min
216 } else {
217 stats.min_value
218 };
219 let max = if old_format {
221 stats.max
222 } else {
223 stats.max_value
224 };
225
226 fn check_len(min: &Option<&[u8]>, max: &Option<&[u8]>, len: usize) -> Result<()> {
227 if let Some(min) = min {
228 if min.len() < len {
229 return Err(general_err!("Insufficient bytes to parse min statistic",));
230 }
231 }
232 if let Some(max) = max {
233 if max.len() < len {
234 return Err(general_err!("Insufficient bytes to parse max statistic",));
235 }
236 }
237 Ok(())
238 }
239
240 let physical_type = column_descr.physical_type();
241 match physical_type {
242 Type::BOOLEAN => check_len(&min, &max, 1),
243 Type::INT32 | Type::FLOAT => check_len(&min, &max, 4),
244 Type::INT64 | Type::DOUBLE => check_len(&min, &max, 8),
245 Type::INT96 => check_len(&min, &max, 12),
246 _ => Ok(()),
247 }?;
248
249 let res = match physical_type {
254 Type::BOOLEAN => FStatistics::boolean(
255 min.map(|data| data[0] != 0),
256 max.map(|data| data[0] != 0),
257 distinct_count,
258 null_count,
259 old_format,
260 ),
261 Type::INT32 => FStatistics::int32(
262 min.map(|data| i32::from_le_bytes(data[..4].try_into().unwrap())),
263 max.map(|data| i32::from_le_bytes(data[..4].try_into().unwrap())),
264 distinct_count,
265 null_count,
266 old_format,
267 ),
268 Type::INT64 => FStatistics::int64(
269 min.map(|data| i64::from_le_bytes(data[..8].try_into().unwrap())),
270 max.map(|data| i64::from_le_bytes(data[..8].try_into().unwrap())),
271 distinct_count,
272 null_count,
273 old_format,
274 ),
275 Type::INT96 => {
276 let min = if let Some(data) = min {
278 assert_eq!(data.len(), 12);
279 Some(Int96::try_from_le_slice(data)?)
280 } else {
281 None
282 };
283 let max = if let Some(data) = max {
284 assert_eq!(data.len(), 12);
285 Some(Int96::try_from_le_slice(data)?)
286 } else {
287 None
288 };
289 FStatistics::int96(min, max, distinct_count, null_count, old_format)
290 }
291 Type::FLOAT => FStatistics::float(
292 min.map(|data| f32::from_le_bytes(data[..4].try_into().unwrap())),
293 max.map(|data| f32::from_le_bytes(data[..4].try_into().unwrap())),
294 distinct_count,
295 null_count,
296 old_format,
297 ),
298 Type::DOUBLE => FStatistics::double(
299 min.map(|data| f64::from_le_bytes(data[..8].try_into().unwrap())),
300 max.map(|data| f64::from_le_bytes(data[..8].try_into().unwrap())),
301 distinct_count,
302 null_count,
303 old_format,
304 ),
305 Type::BYTE_ARRAY => FStatistics::ByteArray(
306 ValueStatistics::new(
307 min.map(ByteArray::from),
308 max.map(ByteArray::from),
309 distinct_count,
310 null_count,
311 old_format,
312 )
313 .with_max_is_exact(stats.is_max_value_exact.unwrap_or(false))
314 .with_min_is_exact(stats.is_min_value_exact.unwrap_or(false)),
315 ),
316 Type::FIXED_LEN_BYTE_ARRAY => FStatistics::FixedLenByteArray(
317 ValueStatistics::new(
318 min.map(ByteArray::from).map(FixedLenByteArray::from),
319 max.map(ByteArray::from).map(FixedLenByteArray::from),
320 distinct_count,
321 null_count,
322 old_format,
323 )
324 .with_max_is_exact(stats.is_max_value_exact.unwrap_or(false))
325 .with_min_is_exact(stats.is_min_value_exact.unwrap_or(false)),
326 ),
327 };
328
329 Some(res)
330 }
331 None => None,
332 })
333}
334
335const COL_META_TYPE: u16 = 1 << 1;
337const COL_META_ENCODINGS: u16 = 1 << 2;
338const COL_META_CODEC: u16 = 1 << 4;
339const COL_META_NUM_VALUES: u16 = 1 << 5;
340const COL_META_TOTAL_UNCOMP_SZ: u16 = 1 << 6;
341const COL_META_TOTAL_COMP_SZ: u16 = 1 << 7;
342const COL_META_DATA_PAGE_OFFSET: u16 = 1 << 9;
343
344const COL_META_ALL_REQUIRED: u16 = COL_META_TYPE
346 | COL_META_ENCODINGS
347 | COL_META_CODEC
348 | COL_META_NUM_VALUES
349 | COL_META_TOTAL_UNCOMP_SZ
350 | COL_META_TOTAL_COMP_SZ
351 | COL_META_DATA_PAGE_OFFSET;
352
353fn validate_column_metadata(mask: u16) -> Result<()> {
356 if mask != COL_META_ALL_REQUIRED {
357 if mask & COL_META_ENCODINGS == 0 {
358 return Err(general_err!("Required field encodings is missing"));
359 }
360
361 if mask & COL_META_CODEC == 0 {
362 return Err(general_err!("Required field codec is missing"));
363 }
364 if mask & COL_META_NUM_VALUES == 0 {
365 return Err(general_err!("Required field num_values is missing"));
366 }
367 if mask & COL_META_TOTAL_UNCOMP_SZ == 0 {
368 return Err(general_err!(
369 "Required field total_uncompressed_size is missing"
370 ));
371 }
372 if mask & COL_META_TOTAL_COMP_SZ == 0 {
373 return Err(general_err!(
374 "Required field total_compressed_size is missing"
375 ));
376 }
377 if mask & COL_META_DATA_PAGE_OFFSET == 0 {
378 return Err(general_err!("Required field data_page_offset is missing"));
379 }
380 }
381
382 Ok(())
383}
384
385fn read_column_metadata<'a>(
388 prot: &mut ThriftSliceInputProtocol<'a>,
389 column: &mut ColumnChunkMetaData,
390) -> Result<u16> {
391 let mut seen_mask = 0u16;
393
394 let column_descr = &column.column_descr;
414
415 let mut last_field_id = 0i16;
416 loop {
417 let field_ident = prot.read_field_begin(last_field_id)?;
418 if field_ident.field_type == FieldType::Stop {
419 break;
420 }
421 match field_ident.id {
422 1 => {
424 Type::read_thrift(&mut *prot)?;
426 seen_mask |= COL_META_TYPE;
427 }
428 2 => {
429 column.encodings = EncodingMask::read_thrift(&mut *prot)?;
430 seen_mask |= COL_META_ENCODINGS;
431 }
432 4 => {
434 column.compression = Compression::read_thrift(&mut *prot)?;
435 seen_mask |= COL_META_CODEC;
436 }
437 5 => {
438 column.num_values = i64::read_thrift(&mut *prot)?;
439 seen_mask |= COL_META_NUM_VALUES;
440 }
441 6 => {
442 column.total_uncompressed_size = i64::read_thrift(&mut *prot)?;
443 seen_mask |= COL_META_TOTAL_UNCOMP_SZ;
444 }
445 7 => {
446 column.total_compressed_size = i64::read_thrift(&mut *prot)?;
447 seen_mask |= COL_META_TOTAL_COMP_SZ;
448 }
449 9 => {
451 column.data_page_offset = i64::read_thrift(&mut *prot)?;
452 seen_mask |= COL_META_DATA_PAGE_OFFSET;
453 }
454 10 => {
455 column.index_page_offset = Some(i64::read_thrift(&mut *prot)?);
456 }
457 11 => {
458 column.dictionary_page_offset = Some(i64::read_thrift(&mut *prot)?);
459 }
460 12 => {
461 column.statistics =
462 convert_stats(column_descr, Some(Statistics::read_thrift(&mut *prot)?))?;
463 }
464 13 => {
465 let val =
466 read_thrift_vec::<PageEncodingStats, ThriftSliceInputProtocol>(&mut *prot)?;
467 column.encoding_stats = Some(val);
468 }
469 14 => {
470 column.bloom_filter_offset = Some(i64::read_thrift(&mut *prot)?);
471 }
472 15 => {
473 column.bloom_filter_length = Some(i32::read_thrift(&mut *prot)?);
474 }
475 16 => {
476 let val = SizeStatistics::read_thrift(&mut *prot)?;
477 column.unencoded_byte_array_data_bytes = val.unencoded_byte_array_data_bytes;
478 column.repetition_level_histogram =
479 val.repetition_level_histogram.map(LevelHistogram::from);
480 column.definition_level_histogram =
481 val.definition_level_histogram.map(LevelHistogram::from);
482 }
483 17 => {
484 let val = GeospatialStatistics::read_thrift(&mut *prot)?;
485 column.geo_statistics = convert_geo_stats(Some(val));
486 }
487 _ => {
488 prot.skip(field_ident.field_type)?;
489 }
490 };
491 last_field_id = field_ident.id;
492 }
493
494 Ok(seen_mask)
495}
496
497fn read_column_chunk<'a>(
500 prot: &mut ThriftSliceInputProtocol<'a>,
501 column_descr: &Arc<ColumnDescriptor>,
502) -> Result<ColumnChunkMetaData> {
503 let mut col = ColumnChunkMetaDataBuilder::new(column_descr.clone()).build()?;
505
506 let mut has_file_offset = false;
508
509 let mut col_meta_mask = 0u16;
511
512 let mut last_field_id = 0i16;
524 loop {
525 let field_ident = prot.read_field_begin(last_field_id)?;
526 if field_ident.field_type == FieldType::Stop {
527 break;
528 }
529 match field_ident.id {
530 1 => {
531 col.file_path = Some(String::read_thrift(&mut *prot)?);
532 }
533 2 => {
534 col.file_offset = i64::read_thrift(&mut *prot)?;
535 has_file_offset = true;
536 }
537 3 => {
538 col_meta_mask = read_column_metadata(&mut *prot, &mut col)?;
539 }
540 4 => {
541 col.offset_index_offset = Some(i64::read_thrift(&mut *prot)?);
542 }
543 5 => {
544 col.offset_index_length = Some(i32::read_thrift(&mut *prot)?);
545 }
546 6 => {
547 col.column_index_offset = Some(i64::read_thrift(&mut *prot)?);
548 }
549 7 => {
550 col.column_index_length = Some(i32::read_thrift(&mut *prot)?);
551 }
552 #[cfg(feature = "encryption")]
553 8 => {
554 let val = ColumnCryptoMetaData::read_thrift(&mut *prot)?;
555 col.column_crypto_metadata = Some(Box::new(val));
556 }
557 #[cfg(feature = "encryption")]
558 9 => {
559 col.encrypted_column_metadata = Some(<&[u8]>::read_thrift(&mut *prot)?.to_vec());
560 }
561 _ => {
562 prot.skip(field_ident.field_type)?;
563 }
564 };
565 last_field_id = field_ident.id;
566 }
567
568 if !has_file_offset {
570 return Err(general_err!("Required field file_offset is missing"));
571 };
572
573 #[cfg(feature = "encryption")]
575 if col.encrypted_column_metadata.is_some() {
576 return Ok(col);
577 }
578
579 validate_column_metadata(col_meta_mask)?;
581
582 Ok(col)
583}
584
585fn read_row_group(
586 prot: &mut ThriftSliceInputProtocol,
587 schema_descr: &Arc<SchemaDescriptor>,
588) -> Result<RowGroupMetaData> {
589 let mut row_group = RowGroupMetaDataBuilder::new(schema_descr.clone()).build_unchecked();
591
592 const RG_COLUMNS: u8 = 1 << 1;
594 const RG_TOT_BYTE_SIZE: u8 = 1 << 2;
595 const RG_NUM_ROWS: u8 = 1 << 3;
596 const RG_ALL_REQUIRED: u8 = RG_COLUMNS | RG_TOT_BYTE_SIZE | RG_NUM_ROWS;
597
598 let mut mask = 0u8;
599
600 let mut last_field_id = 0i16;
610 loop {
611 let field_ident = prot.read_field_begin(last_field_id)?;
612 if field_ident.field_type == FieldType::Stop {
613 break;
614 }
615 match field_ident.id {
616 1 => {
617 let list_ident = prot.read_list_begin()?;
618 if schema_descr.num_columns() != list_ident.size as usize {
619 return Err(general_err!(
620 "Column count mismatch. Schema has {} columns while Row Group has {}",
621 schema_descr.num_columns(),
622 list_ident.size
623 ));
624 }
625 for i in 0..list_ident.size as usize {
626 let col = read_column_chunk(prot, &schema_descr.columns()[i])?;
627 row_group.columns.push(col);
628 }
629 mask |= RG_COLUMNS;
630 }
631 2 => {
632 row_group.total_byte_size = i64::read_thrift(&mut *prot)?;
633 mask |= RG_TOT_BYTE_SIZE;
634 }
635 3 => {
636 row_group.num_rows = i64::read_thrift(&mut *prot)?;
637 mask |= RG_NUM_ROWS;
638 }
639 4 => {
640 let val = read_thrift_vec::<SortingColumn, ThriftSliceInputProtocol>(&mut *prot)?;
641 row_group.sorting_columns = Some(val);
642 }
643 5 => {
644 row_group.file_offset = Some(i64::read_thrift(&mut *prot)?);
645 }
646 7 => {
648 row_group.ordinal = Some(i16::read_thrift(&mut *prot)?);
649 }
650 _ => {
651 prot.skip(field_ident.field_type)?;
652 }
653 };
654 last_field_id = field_ident.id;
655 }
656
657 if mask != RG_ALL_REQUIRED {
658 if mask & RG_COLUMNS == 0 {
659 return Err(general_err!("Required field columns is missing"));
660 }
661 if mask & RG_TOT_BYTE_SIZE == 0 {
662 return Err(general_err!("Required field total_byte_size is missing"));
663 }
664 if mask & RG_NUM_ROWS == 0 {
665 return Err(general_err!("Required field num_rows is missing"));
666 }
667 }
668
669 Ok(row_group)
670}
671
672pub(crate) fn parquet_metadata_from_bytes(buf: &[u8]) -> Result<ParquetMetaData> {
675 let mut prot = ThriftSliceInputProtocol::new(buf);
676
677 let mut version: Option<i32> = None;
679 let mut num_rows: Option<i64> = None;
680 let mut row_groups: Option<Vec<RowGroupMetaData>> = None;
681 let mut key_value_metadata: Option<Vec<KeyValue>> = None;
682 let mut created_by: Option<&str> = None;
683 let mut column_orders: Option<Vec<ColumnOrder>> = None;
684 #[cfg(feature = "encryption")]
685 let mut encryption_algorithm: Option<EncryptionAlgorithm> = None;
686 #[cfg(feature = "encryption")]
687 let mut footer_signing_key_metadata: Option<&[u8]> = None;
688
689 let mut schema_descr: Option<Arc<SchemaDescriptor>> = None;
691
692 let mut last_field_id = 0i16;
704 loop {
705 let field_ident = prot.read_field_begin(last_field_id)?;
706 if field_ident.field_type == FieldType::Stop {
707 break;
708 }
709 match field_ident.id {
710 1 => {
711 version = Some(i32::read_thrift(&mut prot)?);
712 }
713 2 => {
714 let val = read_thrift_vec::<SchemaElement, ThriftSliceInputProtocol>(&mut prot)?;
716 let val = parquet_schema_from_array(val)?;
717 schema_descr = Some(Arc::new(SchemaDescriptor::new(val)));
718 }
719 3 => {
720 num_rows = Some(i64::read_thrift(&mut prot)?);
721 }
722 4 => {
723 if schema_descr.is_none() {
724 return Err(general_err!("Required field schema is missing"));
725 }
726 let schema_descr = schema_descr.as_ref().unwrap();
727 let list_ident = prot.read_list_begin()?;
728 let mut rg_vec = Vec::with_capacity(list_ident.size as usize);
729 for _ in 0..list_ident.size {
730 rg_vec.push(read_row_group(&mut prot, schema_descr)?);
731 }
732 row_groups = Some(rg_vec);
733 }
734 5 => {
735 let val = read_thrift_vec::<KeyValue, ThriftSliceInputProtocol>(&mut prot)?;
736 key_value_metadata = Some(val);
737 }
738 6 => {
739 created_by = Some(<&str>::read_thrift(&mut prot)?);
740 }
741 7 => {
742 let val = read_thrift_vec::<ColumnOrder, ThriftSliceInputProtocol>(&mut prot)?;
743 column_orders = Some(val);
744 }
745 #[cfg(feature = "encryption")]
746 8 => {
747 let val = EncryptionAlgorithm::read_thrift(&mut prot)?;
748 encryption_algorithm = Some(val);
749 }
750 #[cfg(feature = "encryption")]
751 9 => {
752 footer_signing_key_metadata = Some(<&[u8]>::read_thrift(&mut prot)?);
753 }
754 _ => {
755 prot.skip(field_ident.field_type)?;
756 }
757 };
758 last_field_id = field_ident.id;
759 }
760 let Some(version) = version else {
761 return Err(general_err!("Required field version is missing"));
762 };
763 let Some(num_rows) = num_rows else {
764 return Err(general_err!("Required field num_rows is missing"));
765 };
766 let Some(row_groups) = row_groups else {
767 return Err(general_err!("Required field row_groups is missing"));
768 };
769
770 let created_by = created_by.map(|c| c.to_owned());
771
772 let schema_descr = schema_descr.unwrap();
774
775 if column_orders
777 .as_ref()
778 .is_some_and(|cos| cos.len() != schema_descr.num_columns())
779 {
780 return Err(general_err!("Column order length mismatch"));
781 }
782 let column_orders = column_orders.map(|mut cos| {
785 for (i, column) in schema_descr.columns().iter().enumerate() {
786 if let ColumnOrder::TYPE_DEFINED_ORDER(_) = cos[i] {
787 let sort_order = ColumnOrder::get_sort_order(
788 column.logical_type(),
789 column.converted_type(),
790 column.physical_type(),
791 );
792 cos[i] = ColumnOrder::TYPE_DEFINED_ORDER(sort_order);
793 }
794 }
795 cos
796 });
797
798 #[cfg(not(feature = "encryption"))]
799 let fmd = crate::file::metadata::FileMetaData::new(
800 version,
801 num_rows,
802 created_by,
803 key_value_metadata,
804 schema_descr,
805 column_orders,
806 );
807 #[cfg(feature = "encryption")]
808 let fmd = crate::file::metadata::FileMetaData::new(
809 version,
810 num_rows,
811 created_by,
812 key_value_metadata,
813 schema_descr,
814 column_orders,
815 )
816 .with_encryption_algorithm(encryption_algorithm)
817 .with_footer_signing_key_metadata(footer_signing_key_metadata.map(|v| v.to_vec()));
818
819 Ok(ParquetMetaData::new(fmd, row_groups))
820}
821
822thrift_struct!(
823 pub(crate) struct IndexPageHeader {}
824);
825
826thrift_struct!(
827pub(crate) struct DictionaryPageHeader {
828 1: required i32 num_values;
830
831 2: required Encoding encoding
833
834 3: optional bool is_sorted;
836}
837);
838
839thrift_struct!(
840pub(crate) struct PageStatistics {
848 1: optional binary max;
849 2: optional binary min;
850 3: optional i64 null_count;
851 4: optional i64 distinct_count;
852 5: optional binary max_value;
853 6: optional binary min_value;
854 7: optional bool is_max_value_exact;
855 8: optional bool is_min_value_exact;
856}
857);
858
859thrift_struct!(
860pub(crate) struct DataPageHeader {
861 1: required i32 num_values
862 2: required Encoding encoding
863 3: required Encoding definition_level_encoding;
864 4: required Encoding repetition_level_encoding;
865 5: optional PageStatistics statistics;
866}
867);
868
869impl DataPageHeader {
870 fn read_thrift_without_stats<'a, R>(prot: &mut R) -> Result<Self>
872 where
873 R: ThriftCompactInputProtocol<'a>,
874 {
875 let mut num_values: Option<i32> = None;
876 let mut encoding: Option<Encoding> = None;
877 let mut definition_level_encoding: Option<Encoding> = None;
878 let mut repetition_level_encoding: Option<Encoding> = None;
879 let statistics: Option<PageStatistics> = None;
880 let mut last_field_id = 0i16;
881 loop {
882 let field_ident = prot.read_field_begin(last_field_id)?;
883 if field_ident.field_type == FieldType::Stop {
884 break;
885 }
886 match field_ident.id {
887 1 => {
888 let val = i32::read_thrift(&mut *prot)?;
889 num_values = Some(val);
890 }
891 2 => {
892 let val = Encoding::read_thrift(&mut *prot)?;
893 encoding = Some(val);
894 }
895 3 => {
896 let val = Encoding::read_thrift(&mut *prot)?;
897 definition_level_encoding = Some(val);
898 }
899 4 => {
900 let val = Encoding::read_thrift(&mut *prot)?;
901 repetition_level_encoding = Some(val);
902 }
903 _ => {
904 prot.skip(field_ident.field_type)?;
905 }
906 };
907 last_field_id = field_ident.id;
908 }
909 let Some(num_values) = num_values else {
910 return Err(general_err!("Required field num_values is missing"));
911 };
912 let Some(encoding) = encoding else {
913 return Err(general_err!("Required field encoding is missing"));
914 };
915 let Some(definition_level_encoding) = definition_level_encoding else {
916 return Err(general_err!(
917 "Required field definition_level_encoding is missing"
918 ));
919 };
920 let Some(repetition_level_encoding) = repetition_level_encoding else {
921 return Err(general_err!(
922 "Required field repetition_level_encoding is missing"
923 ));
924 };
925 Ok(Self {
926 num_values,
927 encoding,
928 definition_level_encoding,
929 repetition_level_encoding,
930 statistics,
931 })
932 }
933}
934
935thrift_struct!(
936pub(crate) struct DataPageHeaderV2 {
937 1: required i32 num_values
938 2: required i32 num_nulls
939 3: required i32 num_rows
940 4: required Encoding encoding
941 5: required i32 definition_levels_byte_length;
942 6: required i32 repetition_levels_byte_length;
943 7: optional bool is_compressed = true;
944 8: optional PageStatistics statistics;
945}
946);
947
948impl DataPageHeaderV2 {
949 fn read_thrift_without_stats<'a, R>(prot: &mut R) -> Result<Self>
951 where
952 R: ThriftCompactInputProtocol<'a>,
953 {
954 let mut num_values: Option<i32> = None;
955 let mut num_nulls: Option<i32> = None;
956 let mut num_rows: Option<i32> = None;
957 let mut encoding: Option<Encoding> = None;
958 let mut definition_levels_byte_length: Option<i32> = None;
959 let mut repetition_levels_byte_length: Option<i32> = None;
960 let mut is_compressed: Option<bool> = None;
961 let statistics: Option<PageStatistics> = None;
962 let mut last_field_id = 0i16;
963 loop {
964 let field_ident = prot.read_field_begin(last_field_id)?;
965 if field_ident.field_type == FieldType::Stop {
966 break;
967 }
968 match field_ident.id {
969 1 => {
970 let val = i32::read_thrift(&mut *prot)?;
971 num_values = Some(val);
972 }
973 2 => {
974 let val = i32::read_thrift(&mut *prot)?;
975 num_nulls = Some(val);
976 }
977 3 => {
978 let val = i32::read_thrift(&mut *prot)?;
979 num_rows = Some(val);
980 }
981 4 => {
982 let val = Encoding::read_thrift(&mut *prot)?;
983 encoding = Some(val);
984 }
985 5 => {
986 let val = i32::read_thrift(&mut *prot)?;
987 definition_levels_byte_length = Some(val);
988 }
989 6 => {
990 let val = i32::read_thrift(&mut *prot)?;
991 repetition_levels_byte_length = Some(val);
992 }
993 7 => {
994 let val = field_ident.bool_val.unwrap();
995 is_compressed = Some(val);
996 }
997 _ => {
998 prot.skip(field_ident.field_type)?;
999 }
1000 };
1001 last_field_id = field_ident.id;
1002 }
1003 let Some(num_values) = num_values else {
1004 return Err(general_err!("Required field num_values is missing"));
1005 };
1006 let Some(num_nulls) = num_nulls else {
1007 return Err(general_err!("Required field num_nulls is missing"));
1008 };
1009 let Some(num_rows) = num_rows else {
1010 return Err(general_err!("Required field num_rows is missing"));
1011 };
1012 let Some(encoding) = encoding else {
1013 return Err(general_err!("Required field encoding is missing"));
1014 };
1015 let Some(definition_levels_byte_length) = definition_levels_byte_length else {
1016 return Err(general_err!(
1017 "Required field definition_levels_byte_length is missing"
1018 ));
1019 };
1020 let Some(repetition_levels_byte_length) = repetition_levels_byte_length else {
1021 return Err(general_err!(
1022 "Required field repetition_levels_byte_length is missing"
1023 ));
1024 };
1025 Ok(Self {
1026 num_values,
1027 num_nulls,
1028 num_rows,
1029 encoding,
1030 definition_levels_byte_length,
1031 repetition_levels_byte_length,
1032 is_compressed,
1033 statistics,
1034 })
1035 }
1036}
1037
1038thrift_struct!(
1039pub(crate) struct PageHeader {
1040 1: required PageType r#type
1042
1043 2: required i32 uncompressed_page_size
1045
1046 3: required i32 compressed_page_size
1048
1049 4: optional i32 crc
1051
1052 5: optional DataPageHeader data_page_header;
1054 6: optional IndexPageHeader index_page_header;
1055 7: optional DictionaryPageHeader dictionary_page_header;
1056 8: optional DataPageHeaderV2 data_page_header_v2;
1057}
1058);
1059
1060impl PageHeader {
1061 pub(crate) fn read_thrift_without_stats<'a, R>(prot: &mut R) -> Result<Self>
1065 where
1066 R: ThriftCompactInputProtocol<'a>,
1067 {
1068 let mut type_: Option<PageType> = None;
1069 let mut uncompressed_page_size: Option<i32> = None;
1070 let mut compressed_page_size: Option<i32> = None;
1071 let mut crc: Option<i32> = None;
1072 let mut data_page_header: Option<DataPageHeader> = None;
1073 let mut index_page_header: Option<IndexPageHeader> = None;
1074 let mut dictionary_page_header: Option<DictionaryPageHeader> = None;
1075 let mut data_page_header_v2: Option<DataPageHeaderV2> = None;
1076 let mut last_field_id = 0i16;
1077 loop {
1078 let field_ident = prot.read_field_begin(last_field_id)?;
1079 if field_ident.field_type == FieldType::Stop {
1080 break;
1081 }
1082 match field_ident.id {
1083 1 => {
1084 let val = PageType::read_thrift(&mut *prot)?;
1085 type_ = Some(val);
1086 }
1087 2 => {
1088 let val = i32::read_thrift(&mut *prot)?;
1089 uncompressed_page_size = Some(val);
1090 }
1091 3 => {
1092 let val = i32::read_thrift(&mut *prot)?;
1093 compressed_page_size = Some(val);
1094 }
1095 4 => {
1096 let val = i32::read_thrift(&mut *prot)?;
1097 crc = Some(val);
1098 }
1099 5 => {
1100 let val = DataPageHeader::read_thrift_without_stats(&mut *prot)?;
1101 data_page_header = Some(val);
1102 }
1103 6 => {
1104 let val = IndexPageHeader::read_thrift(&mut *prot)?;
1105 index_page_header = Some(val);
1106 }
1107 7 => {
1108 let val = DictionaryPageHeader::read_thrift(&mut *prot)?;
1109 dictionary_page_header = Some(val);
1110 }
1111 8 => {
1112 let val = DataPageHeaderV2::read_thrift_without_stats(&mut *prot)?;
1113 data_page_header_v2 = Some(val);
1114 }
1115 _ => {
1116 prot.skip(field_ident.field_type)?;
1117 }
1118 };
1119 last_field_id = field_ident.id;
1120 }
1121 let Some(type_) = type_ else {
1122 return Err(general_err!("Required field type_ is missing"));
1123 };
1124 let Some(uncompressed_page_size) = uncompressed_page_size else {
1125 return Err(general_err!(
1126 "Required field uncompressed_page_size is missing"
1127 ));
1128 };
1129 let Some(compressed_page_size) = compressed_page_size else {
1130 return Err(general_err!(
1131 "Required field compressed_page_size is missing"
1132 ));
1133 };
1134 Ok(Self {
1135 r#type: type_,
1136 uncompressed_page_size,
1137 compressed_page_size,
1138 crc,
1139 data_page_header,
1140 index_page_header,
1141 dictionary_page_header,
1142 data_page_header_v2,
1143 })
1144 }
1145}
1146
1147pub(super) fn serialize_column_meta_data<W: Write>(
1171 column_chunk: &ColumnChunkMetaData,
1172 w: &mut ThriftCompactOutputProtocol<W>,
1173) -> Result<()> {
1174 use crate::file::statistics::page_stats_to_thrift;
1175
1176 column_chunk.column_type().write_thrift_field(w, 1, 0)?;
1177 column_chunk
1178 .encodings()
1179 .collect::<Vec<_>>()
1180 .write_thrift_field(w, 2, 1)?;
1181 let path = column_chunk.column_descr.path().parts();
1182 let path: Vec<&str> = path.iter().map(|v| v.as_str()).collect();
1183 path.write_thrift_field(w, 3, 2)?;
1184 column_chunk.compression.write_thrift_field(w, 4, 3)?;
1185 column_chunk.num_values.write_thrift_field(w, 5, 4)?;
1186 column_chunk
1187 .total_uncompressed_size
1188 .write_thrift_field(w, 6, 5)?;
1189 column_chunk
1190 .total_compressed_size
1191 .write_thrift_field(w, 7, 6)?;
1192 let mut last_field_id = column_chunk.data_page_offset.write_thrift_field(w, 9, 7)?;
1194 if let Some(index_page_offset) = column_chunk.index_page_offset {
1195 last_field_id = index_page_offset.write_thrift_field(w, 10, last_field_id)?;
1196 }
1197 if let Some(dictionary_page_offset) = column_chunk.dictionary_page_offset {
1198 last_field_id = dictionary_page_offset.write_thrift_field(w, 11, last_field_id)?;
1199 }
1200 let stats = page_stats_to_thrift(column_chunk.statistics());
1202 if let Some(stats) = stats {
1203 last_field_id = stats.write_thrift_field(w, 12, last_field_id)?;
1204 }
1205 if let Some(page_encoding_stats) = column_chunk.page_encoding_stats() {
1206 last_field_id = page_encoding_stats.write_thrift_field(w, 13, last_field_id)?;
1207 }
1208 if let Some(bloom_filter_offset) = column_chunk.bloom_filter_offset {
1209 last_field_id = bloom_filter_offset.write_thrift_field(w, 14, last_field_id)?;
1210 }
1211 if let Some(bloom_filter_length) = column_chunk.bloom_filter_length {
1212 last_field_id = bloom_filter_length.write_thrift_field(w, 15, last_field_id)?;
1213 }
1214
1215 let size_stats = if column_chunk.unencoded_byte_array_data_bytes.is_some()
1217 || column_chunk.repetition_level_histogram.is_some()
1218 || column_chunk.definition_level_histogram.is_some()
1219 {
1220 let repetition_level_histogram = column_chunk
1221 .repetition_level_histogram()
1222 .map(|hist| hist.clone().into_inner());
1223
1224 let definition_level_histogram = column_chunk
1225 .definition_level_histogram()
1226 .map(|hist| hist.clone().into_inner());
1227
1228 Some(SizeStatistics {
1229 unencoded_byte_array_data_bytes: column_chunk.unencoded_byte_array_data_bytes,
1230 repetition_level_histogram,
1231 definition_level_histogram,
1232 })
1233 } else {
1234 None
1235 };
1236 if let Some(size_stats) = size_stats {
1237 last_field_id = size_stats.write_thrift_field(w, 16, last_field_id)?;
1238 }
1239
1240 if let Some(geo_stats) = column_chunk.geo_statistics() {
1241 geo_stats.write_thrift_field(w, 17, last_field_id)?;
1242 }
1243
1244 w.write_struct_end()
1245}
1246
1247pub(super) struct FileMeta<'a> {
1249 pub(super) file_metadata: &'a crate::file::metadata::FileMetaData,
1250 pub(super) row_groups: &'a Vec<RowGroupMetaData>,
1251}
1252
1253impl<'a> WriteThrift for FileMeta<'a> {
1265 const ELEMENT_TYPE: ElementType = ElementType::Struct;
1266
1267 #[allow(unused_assignments)]
1269 fn write_thrift<W: Write>(&self, writer: &mut ThriftCompactOutputProtocol<W>) -> Result<()> {
1270 self.file_metadata
1271 .version
1272 .write_thrift_field(writer, 1, 0)?;
1273
1274 let root = self.file_metadata.schema_descr().root_schema_ptr();
1277 let schema_len = num_nodes(&root)?;
1278 writer.write_field_begin(FieldType::List, 2, 1)?;
1279 writer.write_list_begin(ElementType::Struct, schema_len)?;
1280 write_schema(&root, writer)?;
1282
1283 self.file_metadata
1284 .num_rows
1285 .write_thrift_field(writer, 3, 2)?;
1286
1287 let mut last_field_id = self.row_groups.write_thrift_field(writer, 4, 3)?;
1289
1290 if let Some(kv_metadata) = self.file_metadata.key_value_metadata() {
1291 last_field_id = kv_metadata.write_thrift_field(writer, 5, last_field_id)?;
1292 }
1293 if let Some(created_by) = self.file_metadata.created_by() {
1294 last_field_id = created_by.write_thrift_field(writer, 6, last_field_id)?;
1295 }
1296 if let Some(column_orders) = self.file_metadata.column_orders() {
1297 last_field_id = column_orders.write_thrift_field(writer, 7, last_field_id)?;
1298 }
1299 #[cfg(feature = "encryption")]
1300 if let Some(algo) = self.file_metadata.encryption_algorithm.as_ref() {
1301 last_field_id = algo.write_thrift_field(writer, 8, last_field_id)?;
1302 }
1303 #[cfg(feature = "encryption")]
1304 if let Some(key) = self.file_metadata.footer_signing_key_metadata.as_ref() {
1305 key.as_slice()
1306 .write_thrift_field(writer, 9, last_field_id)?;
1307 }
1308
1309 writer.write_struct_end()
1310 }
1311}
1312
1313fn write_schema<W: Write>(
1314 schema: &TypePtr,
1315 writer: &mut ThriftCompactOutputProtocol<W>,
1316) -> Result<()> {
1317 if !schema.is_group() {
1318 return Err(general_err!("Root schema must be Group type"));
1319 }
1320 write_schema_helper(schema, writer)
1321}
1322
1323fn write_schema_helper<W: Write>(
1324 node: &TypePtr,
1325 writer: &mut ThriftCompactOutputProtocol<W>,
1326) -> Result<()> {
1327 match node.as_ref() {
1328 crate::schema::types::Type::PrimitiveType {
1329 basic_info,
1330 physical_type,
1331 type_length,
1332 scale,
1333 precision,
1334 } => {
1335 let element = SchemaElement {
1336 r#type: Some(*physical_type),
1337 type_length: if *type_length >= 0 {
1338 Some(*type_length)
1339 } else {
1340 None
1341 },
1342 repetition_type: Some(basic_info.repetition()),
1343 name: basic_info.name(),
1344 num_children: None,
1345 converted_type: match basic_info.converted_type() {
1346 ConvertedType::NONE => None,
1347 other => Some(other),
1348 },
1349 scale: if *scale >= 0 { Some(*scale) } else { None },
1350 precision: if *precision >= 0 {
1351 Some(*precision)
1352 } else {
1353 None
1354 },
1355 field_id: if basic_info.has_id() {
1356 Some(basic_info.id())
1357 } else {
1358 None
1359 },
1360 logical_type: basic_info.logical_type(),
1361 };
1362 element.write_thrift(writer)
1363 }
1364 crate::schema::types::Type::GroupType { basic_info, fields } => {
1365 let repetition = if basic_info.has_repetition() {
1366 Some(basic_info.repetition())
1367 } else {
1368 None
1369 };
1370
1371 let element = SchemaElement {
1372 r#type: None,
1373 type_length: None,
1374 repetition_type: repetition,
1375 name: basic_info.name(),
1376 num_children: Some(fields.len().try_into()?),
1377 converted_type: match basic_info.converted_type() {
1378 ConvertedType::NONE => None,
1379 other => Some(other),
1380 },
1381 scale: None,
1382 precision: None,
1383 field_id: if basic_info.has_id() {
1384 Some(basic_info.id())
1385 } else {
1386 None
1387 },
1388 logical_type: basic_info.logical_type(),
1389 };
1390
1391 element.write_thrift(writer)?;
1392
1393 for field in fields {
1395 write_schema_helper(field, writer)?;
1396 }
1397 Ok(())
1398 }
1399 }
1400}
1401
1402impl WriteThrift for RowGroupMetaData {
1412 const ELEMENT_TYPE: ElementType = ElementType::Struct;
1413
1414 fn write_thrift<W: Write>(&self, writer: &mut ThriftCompactOutputProtocol<W>) -> Result<()> {
1415 self.columns.write_thrift_field(writer, 1, 0)?;
1417 self.total_byte_size.write_thrift_field(writer, 2, 1)?;
1418 let mut last_field_id = self.num_rows.write_thrift_field(writer, 3, 2)?;
1419 if let Some(sorting_columns) = self.sorting_columns() {
1420 last_field_id = sorting_columns.write_thrift_field(writer, 4, last_field_id)?;
1421 }
1422 if let Some(file_offset) = self.file_offset() {
1423 last_field_id = file_offset.write_thrift_field(writer, 5, last_field_id)?;
1424 }
1425 last_field_id = self
1427 .compressed_size()
1428 .write_thrift_field(writer, 6, last_field_id)?;
1429 if let Some(ordinal) = self.ordinal() {
1430 ordinal.write_thrift_field(writer, 7, last_field_id)?;
1431 }
1432 writer.write_struct_end()
1433 }
1434}
1435
1436impl WriteThrift for ColumnChunkMetaData {
1448 const ELEMENT_TYPE: ElementType = ElementType::Struct;
1449
1450 #[allow(unused_assignments)]
1451 fn write_thrift<W: Write>(&self, writer: &mut ThriftCompactOutputProtocol<W>) -> Result<()> {
1452 let mut last_field_id = 0i16;
1453 if let Some(file_path) = self.file_path() {
1454 last_field_id = file_path.write_thrift_field(writer, 1, last_field_id)?;
1455 }
1456 last_field_id = self
1457 .file_offset()
1458 .write_thrift_field(writer, 2, last_field_id)?;
1459
1460 #[cfg(feature = "encryption")]
1461 {
1462 if self.encrypted_column_metadata.is_none() {
1464 writer.write_field_begin(FieldType::Struct, 3, last_field_id)?;
1465 serialize_column_meta_data(self, writer)?;
1466 last_field_id = 3;
1467 }
1468 }
1469 #[cfg(not(feature = "encryption"))]
1470 {
1471 writer.write_field_begin(FieldType::Struct, 3, last_field_id)?;
1473 serialize_column_meta_data(self, writer)?;
1474 last_field_id = 3;
1475 }
1476
1477 if let Some(offset_idx_off) = self.offset_index_offset() {
1478 last_field_id = offset_idx_off.write_thrift_field(writer, 4, last_field_id)?;
1479 }
1480 if let Some(offset_idx_len) = self.offset_index_length() {
1481 last_field_id = offset_idx_len.write_thrift_field(writer, 5, last_field_id)?;
1482 }
1483 if let Some(column_idx_off) = self.column_index_offset() {
1484 last_field_id = column_idx_off.write_thrift_field(writer, 6, last_field_id)?;
1485 }
1486 if let Some(column_idx_len) = self.column_index_length() {
1487 last_field_id = column_idx_len.write_thrift_field(writer, 7, last_field_id)?;
1488 }
1489 #[cfg(feature = "encryption")]
1490 {
1491 if let Some(crypto_metadata) = self.crypto_metadata() {
1492 last_field_id = crypto_metadata.write_thrift_field(writer, 8, last_field_id)?;
1493 }
1494 if let Some(encrypted_meta) = self.encrypted_column_metadata.as_ref() {
1495 encrypted_meta
1496 .as_slice()
1497 .write_thrift_field(writer, 9, last_field_id)?;
1498 }
1499 }
1500
1501 writer.write_struct_end()
1502 }
1503}
1504
1505impl WriteThrift for crate::geospatial::statistics::GeospatialStatistics {
1510 const ELEMENT_TYPE: ElementType = ElementType::Struct;
1511
1512 fn write_thrift<W: Write>(&self, writer: &mut ThriftCompactOutputProtocol<W>) -> Result<()> {
1513 let mut last_field_id = 0i16;
1514 if let Some(bbox) = self.bounding_box() {
1515 last_field_id = bbox.write_thrift_field(writer, 1, last_field_id)?;
1516 }
1517 if let Some(geo_types) = self.geospatial_types() {
1518 geo_types.write_thrift_field(writer, 2, last_field_id)?;
1519 }
1520
1521 writer.write_struct_end()
1522 }
1523}
1524
1525use crate::geospatial::statistics::GeospatialStatistics as RustGeospatialStatistics;
1527write_thrift_field!(RustGeospatialStatistics, FieldType::Struct);
1528
1529impl WriteThrift for crate::geospatial::bounding_box::BoundingBox {
1540 const ELEMENT_TYPE: ElementType = ElementType::Struct;
1541
1542 fn write_thrift<W: Write>(&self, writer: &mut ThriftCompactOutputProtocol<W>) -> Result<()> {
1543 self.get_xmin().write_thrift_field(writer, 1, 0)?;
1544 self.get_xmax().write_thrift_field(writer, 2, 1)?;
1545 self.get_ymin().write_thrift_field(writer, 3, 2)?;
1546 let mut last_field_id = self.get_ymax().write_thrift_field(writer, 4, 3)?;
1547
1548 if let Some(zmin) = self.get_zmin() {
1549 last_field_id = zmin.write_thrift_field(writer, 5, last_field_id)?;
1550 }
1551 if let Some(zmax) = self.get_zmax() {
1552 last_field_id = zmax.write_thrift_field(writer, 6, last_field_id)?;
1553 }
1554 if let Some(mmin) = self.get_mmin() {
1555 last_field_id = mmin.write_thrift_field(writer, 7, last_field_id)?;
1556 }
1557 if let Some(mmax) = self.get_mmax() {
1558 mmax.write_thrift_field(writer, 8, last_field_id)?;
1559 }
1560
1561 writer.write_struct_end()
1562 }
1563}
1564
1565use crate::geospatial::bounding_box::BoundingBox as RustBoundingBox;
1567write_thrift_field!(RustBoundingBox, FieldType::Struct);
1568
1569#[cfg(test)]
1570pub(crate) mod tests {
1571 use crate::errors::Result;
1572 use crate::file::metadata::thrift::{BoundingBox, SchemaElement, write_schema};
1573 use crate::file::metadata::{ColumnChunkMetaData, RowGroupMetaData};
1574 use crate::parquet_thrift::tests::test_roundtrip;
1575 use crate::parquet_thrift::{
1576 ElementType, ThriftCompactOutputProtocol, ThriftSliceInputProtocol, read_thrift_vec,
1577 };
1578 use crate::schema::types::{
1579 ColumnDescriptor, SchemaDescriptor, TypePtr, num_nodes, parquet_schema_from_array,
1580 };
1581 use std::sync::Arc;
1582
1583 pub(crate) fn read_row_group(
1585 buf: &mut [u8],
1586 schema_descr: Arc<SchemaDescriptor>,
1587 ) -> Result<RowGroupMetaData> {
1588 let mut reader = ThriftSliceInputProtocol::new(buf);
1589 crate::file::metadata::thrift::read_row_group(&mut reader, &schema_descr)
1590 }
1591
1592 pub(crate) fn read_column_chunk(
1593 buf: &mut [u8],
1594 column_descr: Arc<ColumnDescriptor>,
1595 ) -> Result<ColumnChunkMetaData> {
1596 let mut reader = ThriftSliceInputProtocol::new(buf);
1597 crate::file::metadata::thrift::read_column_chunk(&mut reader, &column_descr)
1598 }
1599
1600 pub(crate) fn roundtrip_schema(schema: TypePtr) -> Result<TypePtr> {
1601 let num_nodes = num_nodes(&schema)?;
1602 let mut buf = Vec::new();
1603 let mut writer = ThriftCompactOutputProtocol::new(&mut buf);
1604
1605 writer.write_list_begin(ElementType::Struct, num_nodes)?;
1607
1608 write_schema(&schema, &mut writer)?;
1610
1611 let mut prot = ThriftSliceInputProtocol::new(&buf);
1612 let se: Vec<SchemaElement> = read_thrift_vec(&mut prot)?;
1613 parquet_schema_from_array(se)
1614 }
1615
1616 pub(crate) fn schema_to_buf(schema: &TypePtr) -> Result<Vec<u8>> {
1617 let num_nodes = num_nodes(schema)?;
1618 let mut buf = Vec::new();
1619 let mut writer = ThriftCompactOutputProtocol::new(&mut buf);
1620
1621 writer.write_list_begin(ElementType::Struct, num_nodes)?;
1623
1624 write_schema(schema, &mut writer)?;
1626 Ok(buf)
1627 }
1628
1629 pub(crate) fn buf_to_schema_list<'a>(buf: &'a mut Vec<u8>) -> Result<Vec<SchemaElement<'a>>> {
1630 let mut prot = ThriftSliceInputProtocol::new(buf.as_mut_slice());
1631 read_thrift_vec(&mut prot)
1632 }
1633
1634 #[test]
1635 fn test_bounding_box_roundtrip() {
1636 test_roundtrip(BoundingBox {
1637 xmin: 0.1.into(),
1638 xmax: 10.3.into(),
1639 ymin: 0.001.into(),
1640 ymax: 128.5.into(),
1641 zmin: None,
1642 zmax: None,
1643 mmin: None,
1644 mmax: None,
1645 });
1646
1647 test_roundtrip(BoundingBox {
1648 xmin: 0.1.into(),
1649 xmax: 10.3.into(),
1650 ymin: 0.001.into(),
1651 ymax: 128.5.into(),
1652 zmin: Some(11.0.into()),
1653 zmax: Some(1300.0.into()),
1654 mmin: None,
1655 mmax: None,
1656 });
1657
1658 test_roundtrip(BoundingBox {
1659 xmin: 0.1.into(),
1660 xmax: 10.3.into(),
1661 ymin: 0.001.into(),
1662 ymax: 128.5.into(),
1663 zmin: Some(11.0.into()),
1664 zmax: Some(1300.0.into()),
1665 mmin: Some(3.7.into()),
1666 mmax: Some(42.0.into()),
1667 });
1668 }
1669}