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