1use std::io::Write;
27use std::sync::Arc;
28
29#[cfg(feature = "encryption")]
30pub(crate) mod encryption;
31
32#[cfg(feature = "encryption")]
33use crate::file::{
34 column_crypto_metadata::ColumnCryptoMetaData, metadata::thrift::encryption::EncryptionAlgorithm,
35};
36use crate::{
37 basic::{
38 ColumnOrder, CompressionCodec, ConvertedType, Encoding, EncodingMask, LogicalType,
39 PageType, Repetition, Type,
40 },
41 data_type::{ByteArray, FixedLenByteArray, Int96},
42 errors::{ParquetError, Result},
43 file::{
44 metadata::{
45 ColumnChunkMetaData, ColumnChunkMetaDataBuilder, KeyValue, LevelHistogram,
46 PageEncodingStats, ParquetMetaData, ParquetMetaDataOptions, ParquetPageEncodingStats,
47 RowGroupMetaData, RowGroupMetaDataBuilder, SortingColumn,
48 },
49 statistics::ValueStatistics,
50 },
51 parquet_thrift::{
52 ElementType, FieldType, ReadThrift, ThriftCompactInputProtocol,
53 ThriftCompactOutputProtocol, ThriftSliceInputProtocol, WriteThrift, WriteThriftField,
54 read_thrift_vec, validate_list_type,
55 },
56 schema::types::{
57 ColumnDescriptor, SchemaDescriptor, TypePtr, num_nodes, parquet_schema_from_array,
58 },
59 thrift_struct,
60 util::bit_util::FromBytes,
61 write_thrift_field,
62};
63
64thrift_struct!(
66pub(crate) struct SchemaElement<'a> {
67 1: optional Type r#type;
69 2: optional i32 type_length;
74 3: optional Repetition repetition_type;
77 4: required string<'a> name;
79 5: optional i32 num_children;
84 6: optional ConvertedType converted_type;
89 7: optional i32 scale
94 8: optional i32 precision
95 9: optional i32 field_id;
98 10: optional LogicalType logical_type
103}
104);
105
106thrift_struct!(
107struct Statistics<'a> {
108 1: optional binary<'a> max;
109 2: optional binary<'a> min;
110 3: optional i64 null_count;
111 4: optional i64 distinct_count;
112 5: optional binary<'a> max_value;
113 6: optional binary<'a> min_value;
114 7: optional bool is_max_value_exact;
115 8: optional bool is_min_value_exact;
116 9: optional i64 nan_count;
117}
118);
119
120thrift_struct!(
121struct BoundingBox {
122 1: required double xmin;
123 2: required double xmax;
124 3: required double ymin;
125 4: required double ymax;
126 5: optional double zmin;
127 6: optional double zmax;
128 7: optional double mmin;
129 8: optional double mmax;
130}
131);
132
133thrift_struct!(
134struct GeospatialStatistics {
135 1: optional BoundingBox bbox;
136 2: optional list<i32> geospatial_types;
137}
138);
139
140thrift_struct!(
141struct SizeStatistics {
142 1: optional i64 unencoded_byte_array_data_bytes;
143 2: optional list<i64> repetition_level_histogram;
144 3: optional list<i64> definition_level_histogram;
145}
146);
147
148fn convert_geo_stats(
149 st: GeospatialStatistics,
150) -> crate::geospatial::statistics::GeospatialStatistics {
151 let bbox = st.bbox.map(convert_bounding_box);
152 let geospatial_types: Option<Vec<i32>> = st.geospatial_types.filter(|v| !v.is_empty());
153 crate::geospatial::statistics::GeospatialStatistics::new(bbox, geospatial_types)
154}
155
156fn convert_bounding_box(bb: BoundingBox) -> crate::geospatial::bounding_box::BoundingBox {
157 let mut newbb = crate::geospatial::bounding_box::BoundingBox::new(
158 bb.xmin.into(),
159 bb.xmax.into(),
160 bb.ymin.into(),
161 bb.ymax.into(),
162 );
163
164 newbb = match (bb.zmin, bb.zmax) {
165 (Some(zmin), Some(zmax)) => newbb.with_zrange(zmin.into(), zmax.into()),
166 _ => newbb,
168 };
169
170 newbb = match (bb.mmin, bb.mmax) {
171 (Some(mmin), Some(mmax)) => newbb.with_mrange(mmin.into(), mmax.into()),
172 _ => newbb,
174 };
175
176 newbb
177}
178
179fn convert_stats(
181 column_descr: &Arc<ColumnDescriptor>,
182 thrift_stats: Option<Statistics>,
183) -> Result<Option<crate::file::statistics::Statistics>> {
184 use crate::file::statistics::Statistics as FStatistics;
185 Ok(match thrift_stats {
186 Some(stats) => {
187 let null_count = stats
189 .null_count
190 .map(|null_count| {
191 if null_count < 0 {
192 return Err(general_err!(
193 "Statistics null count is negative {}",
194 null_count
195 ));
196 }
197 Ok(null_count as u64)
198 })
199 .transpose()?;
200 let distinct_count = stats.distinct_count.map(|value| value as u64);
202 let nan_count = stats
204 .nan_count
205 .map(|nan_count| {
206 if nan_count < 0 {
207 return Err(general_err!(
208 "Statistics NaN count is negative {}",
209 nan_count
210 ));
211 }
212 Ok(nan_count as u64)
213 })
214 .transpose()?;
215 let old_format = stats.min_value.is_none() && stats.max_value.is_none();
217 let min = if old_format {
219 stats.min
220 } else {
221 stats.min_value
222 };
223 let max = if old_format {
225 stats.max
226 } else {
227 stats.max_value
228 };
229
230 fn check_len(min: Option<&[u8]>, max: Option<&[u8]>, len: usize) -> Result<()> {
231 if let Some(min) = min
232 && min.len() < len
233 {
234 return Err(general_err!("Insufficient bytes to parse min statistic",));
235 }
236 if let Some(max) = max
237 && max.len() < len
238 {
239 return Err(general_err!("Insufficient bytes to parse max statistic",));
240 }
241 Ok(())
242 }
243
244 let physical_type = column_descr.physical_type();
245 match physical_type {
246 Type::BOOLEAN => check_len(min, max, 1),
247 Type::INT32 | Type::FLOAT => check_len(min, max, 4),
248 Type::INT64 | Type::DOUBLE => check_len(min, max, 8),
249 Type::INT96 => check_len(min, max, 12),
250 _ => Ok(()),
251 }?;
252
253 let res = match physical_type {
258 Type::BOOLEAN => FStatistics::boolean(
259 min.map(|data| data[0] != 0),
260 max.map(|data| data[0] != 0),
261 distinct_count,
262 null_count,
263 old_format,
264 ),
265 Type::INT32 => FStatistics::int32(
266 min.map(|data| i32::from_le_bytes(data[..4].try_into().unwrap())),
267 max.map(|data| i32::from_le_bytes(data[..4].try_into().unwrap())),
268 distinct_count,
269 null_count,
270 old_format,
271 ),
272 Type::INT64 => FStatistics::int64(
273 min.map(|data| i64::from_le_bytes(data[..8].try_into().unwrap())),
274 max.map(|data| i64::from_le_bytes(data[..8].try_into().unwrap())),
275 distinct_count,
276 null_count,
277 old_format,
278 ),
279 Type::INT96 => {
280 let min = if let Some(data) = min {
282 if data.len() != 12 {
283 return Err(general_err!("Incorrect Int96 min statistics"));
284 }
285 Some(Int96::try_from_le_slice(data)?)
286 } else {
287 None
288 };
289 let max = if let Some(data) = max {
290 if data.len() != 12 {
291 return Err(general_err!("Incorrect Int96 max statistics"));
292 }
293 Some(Int96::try_from_le_slice(data)?)
294 } else {
295 None
296 };
297 FStatistics::int96(min, max, distinct_count, null_count, old_format)
298 }
299 Type::FLOAT => FStatistics::Float(
300 ValueStatistics::new(
301 min.map(|data| f32::from_le_bytes(data[..4].try_into().unwrap())),
302 max.map(|data| f32::from_le_bytes(data[..4].try_into().unwrap())),
303 distinct_count,
304 null_count,
305 old_format,
306 )
307 .with_nan_count(nan_count),
308 ),
309 Type::DOUBLE => FStatistics::Double(
310 ValueStatistics::new(
311 min.map(|data| f64::from_le_bytes(data[..8].try_into().unwrap())),
312 max.map(|data| f64::from_le_bytes(data[..8].try_into().unwrap())),
313 distinct_count,
314 null_count,
315 old_format,
316 )
317 .with_nan_count(nan_count),
318 ),
319 Type::BYTE_ARRAY => FStatistics::ByteArray(
320 ValueStatistics::new(
321 min.map(ByteArray::from),
322 max.map(ByteArray::from),
323 distinct_count,
324 null_count,
325 old_format,
326 )
327 .with_max_is_exact(stats.is_max_value_exact.unwrap_or(false))
328 .with_min_is_exact(stats.is_min_value_exact.unwrap_or(false)),
329 ),
330 Type::FIXED_LEN_BYTE_ARRAY => FStatistics::FixedLenByteArray(
331 ValueStatistics::new(
332 min.map(ByteArray::from).map(FixedLenByteArray::from),
333 max.map(ByteArray::from).map(FixedLenByteArray::from),
334 distinct_count,
335 null_count,
336 old_format,
337 )
338 .with_nan_count(nan_count)
339 .with_max_is_exact(stats.is_max_value_exact.unwrap_or(false))
340 .with_min_is_exact(stats.is_min_value_exact.unwrap_or(false)),
341 ),
342 };
343
344 Some(res)
345 }
346 None => None,
347 })
348}
349
350const COL_META_TYPE: u16 = 1 << 1;
352const COL_META_ENCODINGS: u16 = 1 << 2;
353const COL_META_CODEC: u16 = 1 << 4;
354const COL_META_NUM_VALUES: u16 = 1 << 5;
355const COL_META_TOTAL_UNCOMP_SZ: u16 = 1 << 6;
356const COL_META_TOTAL_COMP_SZ: u16 = 1 << 7;
357const COL_META_DATA_PAGE_OFFSET: u16 = 1 << 9;
358
359const COL_META_ALL_REQUIRED: u16 = COL_META_TYPE
361 | COL_META_ENCODINGS
362 | COL_META_CODEC
363 | COL_META_NUM_VALUES
364 | COL_META_TOTAL_UNCOMP_SZ
365 | COL_META_TOTAL_COMP_SZ
366 | COL_META_DATA_PAGE_OFFSET;
367
368fn validate_column_metadata(mask: u16) -> Result<()> {
371 if mask != COL_META_ALL_REQUIRED {
372 if mask & COL_META_ENCODINGS == 0 {
373 return Err(general_err!("Required field encodings is missing"));
374 }
375
376 if mask & COL_META_CODEC == 0 {
377 return Err(general_err!("Required field codec is missing"));
378 }
379 if mask & COL_META_NUM_VALUES == 0 {
380 return Err(general_err!("Required field num_values is missing"));
381 }
382 if mask & COL_META_TOTAL_UNCOMP_SZ == 0 {
383 return Err(general_err!(
384 "Required field total_uncompressed_size is missing"
385 ));
386 }
387 if mask & COL_META_TOTAL_COMP_SZ == 0 {
388 return Err(general_err!(
389 "Required field total_compressed_size is missing"
390 ));
391 }
392 if mask & COL_META_DATA_PAGE_OFFSET == 0 {
393 return Err(general_err!("Required field data_page_offset is missing"));
394 }
395 }
396
397 Ok(())
398}
399
400fn read_encoding_stats_as_mask(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 mut rg_vec = Vec::with_capacity(list_ident.size as usize);
824
825 for _ in 0..list_ident.size {
826 rg_vec.push(read_row_group(&mut prot, schema_descr, options)?);
827 }
828 ensure_row_group_ordinals(&mut rg_vec)?;
829 row_groups = Some(rg_vec);
830 }
831 5 => {
832 let val = read_thrift_vec::<KeyValue, ThriftSliceInputProtocol>(&mut prot)?;
833 key_value_metadata = Some(val);
834 }
835 6 => {
836 created_by = Some(<&str>::read_thrift(&mut prot)?);
837 }
838 7 => {
839 let val = read_thrift_vec::<ColumnOrder, ThriftSliceInputProtocol>(&mut prot)?;
840 column_orders = Some(val);
841 }
842 #[cfg(feature = "encryption")]
843 8 => {
844 let val = EncryptionAlgorithm::read_thrift(&mut prot)?;
845 encryption_algorithm = Some(val);
846 }
847 #[cfg(feature = "encryption")]
848 9 => {
849 footer_signing_key_metadata = Some(<&[u8]>::read_thrift(&mut prot)?);
850 }
851 _ => {
852 prot.skip(field_ident.field_type)?;
853 }
854 }
855 last_field_id = field_ident.id;
856 }
857 let Some(version) = version else {
858 return Err(general_err!("Required field version is missing"));
859 };
860 let Some(num_rows) = num_rows else {
861 return Err(general_err!("Required field num_rows is missing"));
862 };
863 let Some(row_groups) = row_groups else {
864 return Err(general_err!("Required field row_groups is missing"));
865 };
866
867 let created_by = created_by.map(|c| c.to_owned());
868
869 let schema_descr = schema_descr.unwrap();
871
872 if column_orders
874 .as_ref()
875 .is_some_and(|cos| cos.len() != schema_descr.num_columns())
876 {
877 return Err(general_err!("Column order length mismatch"));
878 }
879 let column_orders = column_orders.map(|mut cos| {
881 for (i, column) in schema_descr.columns().iter().enumerate() {
882 if let ColumnOrder::TYPE_DEFINED_ORDER(_) = cos[i] {
883 let sort_order = ColumnOrder::get_sort_order_for_type(
887 column.logical_type_ref(),
888 column.converted_type(),
889 column.physical_type(),
890 true,
891 );
892 cos[i] = ColumnOrder::TYPE_DEFINED_ORDER(sort_order);
893 }
894 }
895 cos
896 });
897
898 #[cfg(not(feature = "encryption"))]
899 let fmd = crate::file::metadata::FileMetaData::new(
900 version,
901 num_rows,
902 created_by,
903 key_value_metadata,
904 schema_descr,
905 column_orders,
906 );
907 #[cfg(feature = "encryption")]
908 let fmd = crate::file::metadata::FileMetaData::new(
909 version,
910 num_rows,
911 created_by,
912 key_value_metadata,
913 schema_descr,
914 column_orders,
915 )
916 .with_encryption_algorithm(encryption_algorithm)
917 .with_footer_signing_key_metadata(footer_signing_key_metadata.map(|v| v.to_vec()));
918
919 Ok(ParquetMetaData::new(fmd, row_groups))
920}
921
922fn ensure_row_group_ordinals(row_groups: &mut [RowGroupMetaData]) -> Result<()> {
939 if row_groups.iter().any(|rg| rg.ordinal.is_some()) {
943 return Ok(());
944 }
945 for (idx, rg) in row_groups.iter_mut().enumerate() {
946 let ordinal: i32 = idx
947 .try_into()
948 .map_err(|_| general_err!("Row group ordinal {} exceeds i32 max value", idx))?;
949 rg.ordinal = Some(ordinal);
950 }
951 Ok(())
952}
953
954thrift_struct!(
955 pub(crate) struct IndexPageHeader {}
956);
957
958thrift_struct!(
959pub(crate) struct DictionaryPageHeader {
960 1: required i32 num_values;
962
963 2: required Encoding encoding
965
966 3: optional bool is_sorted;
968}
969);
970
971thrift_struct!(
972pub(crate) struct PageStatistics {
980 1: optional binary max;
981 2: optional binary min;
982 3: optional i64 null_count;
983 4: optional i64 distinct_count;
984 5: optional binary max_value;
985 6: optional binary min_value;
986 7: optional bool is_max_value_exact;
987 8: optional bool is_min_value_exact;
988 9: optional i64 nan_count;
989}
990);
991
992thrift_struct!(
993pub(crate) struct DataPageHeader {
994 1: required i32 num_values
995 2: required Encoding encoding
996 3: required Encoding definition_level_encoding;
997 4: required Encoding repetition_level_encoding;
998 5: optional PageStatistics statistics;
999}
1000);
1001
1002impl DataPageHeader {
1003 fn read_thrift_without_stats<'a, R>(prot: &mut R) -> Result<Self>
1005 where
1006 R: ThriftCompactInputProtocol<'a>,
1007 {
1008 let mut num_values: Option<i32> = None;
1009 let mut encoding: Option<Encoding> = None;
1010 let mut definition_level_encoding: Option<Encoding> = None;
1011 let mut repetition_level_encoding: Option<Encoding> = None;
1012 let statistics: Option<PageStatistics> = None;
1013 let mut last_field_id = 0i16;
1014 loop {
1015 let field_ident = prot.read_field_begin(last_field_id)?;
1016 if field_ident.field_type == FieldType::Stop {
1017 break;
1018 }
1019 match field_ident.id {
1020 1 => {
1021 let val = i32::read_thrift(&mut *prot)?;
1022 num_values = Some(val);
1023 }
1024 2 => {
1025 let val = Encoding::read_thrift(&mut *prot)?;
1026 encoding = Some(val);
1027 }
1028 3 => {
1029 let val = Encoding::read_thrift(&mut *prot)?;
1030 definition_level_encoding = Some(val);
1031 }
1032 4 => {
1033 let val = Encoding::read_thrift(&mut *prot)?;
1034 repetition_level_encoding = Some(val);
1035 }
1036 _ => {
1037 prot.skip(field_ident.field_type)?;
1038 }
1039 }
1040 last_field_id = field_ident.id;
1041 }
1042 let Some(num_values) = num_values else {
1043 return Err(general_err!("Required field num_values is missing"));
1044 };
1045 let Some(encoding) = encoding else {
1046 return Err(general_err!("Required field encoding is missing"));
1047 };
1048 let Some(definition_level_encoding) = definition_level_encoding else {
1049 return Err(general_err!(
1050 "Required field definition_level_encoding is missing"
1051 ));
1052 };
1053 let Some(repetition_level_encoding) = repetition_level_encoding else {
1054 return Err(general_err!(
1055 "Required field repetition_level_encoding is missing"
1056 ));
1057 };
1058 Ok(Self {
1059 num_values,
1060 encoding,
1061 definition_level_encoding,
1062 repetition_level_encoding,
1063 statistics,
1064 })
1065 }
1066}
1067
1068thrift_struct!(
1069pub(crate) struct DataPageHeaderV2 {
1070 1: required i32 num_values
1071 2: required i32 num_nulls
1072 3: required i32 num_rows
1073 4: required Encoding encoding
1074 5: required i32 definition_levels_byte_length;
1075 6: required i32 repetition_levels_byte_length;
1076 7: optional bool is_compressed = true;
1077 8: optional PageStatistics statistics;
1078}
1079);
1080
1081impl DataPageHeaderV2 {
1082 fn read_thrift_without_stats<'a, R>(prot: &mut R) -> Result<Self>
1084 where
1085 R: ThriftCompactInputProtocol<'a>,
1086 {
1087 let mut num_values: Option<i32> = None;
1088 let mut num_nulls: Option<i32> = None;
1089 let mut num_rows: Option<i32> = None;
1090 let mut encoding: Option<Encoding> = None;
1091 let mut definition_levels_byte_length: Option<i32> = None;
1092 let mut repetition_levels_byte_length: Option<i32> = None;
1093 let mut is_compressed: Option<bool> = None;
1094 let statistics: Option<PageStatistics> = None;
1095 let mut last_field_id = 0i16;
1096 loop {
1097 let field_ident = prot.read_field_begin(last_field_id)?;
1098 if field_ident.field_type == FieldType::Stop {
1099 break;
1100 }
1101 match field_ident.id {
1102 1 => {
1103 let val = i32::read_thrift(&mut *prot)?;
1104 num_values = Some(val);
1105 }
1106 2 => {
1107 let val = i32::read_thrift(&mut *prot)?;
1108 num_nulls = Some(val);
1109 }
1110 3 => {
1111 let val = i32::read_thrift(&mut *prot)?;
1112 num_rows = Some(val);
1113 }
1114 4 => {
1115 let val = Encoding::read_thrift(&mut *prot)?;
1116 encoding = Some(val);
1117 }
1118 5 => {
1119 let val = i32::read_thrift(&mut *prot)?;
1120 definition_levels_byte_length = Some(val);
1121 }
1122 6 => {
1123 let val = i32::read_thrift(&mut *prot)?;
1124 repetition_levels_byte_length = Some(val);
1125 }
1126 7 => {
1127 is_compressed = Some(field_ident.bool_val()?);
1128 }
1129 _ => {
1130 prot.skip(field_ident.field_type)?;
1131 }
1132 }
1133 last_field_id = field_ident.id;
1134 }
1135 let Some(num_values) = num_values else {
1136 return Err(general_err!("Required field num_values is missing"));
1137 };
1138 let Some(num_nulls) = num_nulls else {
1139 return Err(general_err!("Required field num_nulls is missing"));
1140 };
1141 let Some(num_rows) = num_rows else {
1142 return Err(general_err!("Required field num_rows is missing"));
1143 };
1144 let Some(encoding) = encoding else {
1145 return Err(general_err!("Required field encoding is missing"));
1146 };
1147 let Some(definition_levels_byte_length) = definition_levels_byte_length else {
1148 return Err(general_err!(
1149 "Required field definition_levels_byte_length is missing"
1150 ));
1151 };
1152 let Some(repetition_levels_byte_length) = repetition_levels_byte_length else {
1153 return Err(general_err!(
1154 "Required field repetition_levels_byte_length is missing"
1155 ));
1156 };
1157 Ok(Self {
1158 num_values,
1159 num_nulls,
1160 num_rows,
1161 encoding,
1162 definition_levels_byte_length,
1163 repetition_levels_byte_length,
1164 is_compressed,
1165 statistics,
1166 })
1167 }
1168}
1169
1170thrift_struct!(
1171pub(crate) struct PageHeader {
1172 1: required PageType r#type
1174
1175 2: required i32 uncompressed_page_size
1177
1178 3: required i32 compressed_page_size
1180
1181 4: optional i32 crc
1183
1184 5: optional DataPageHeader data_page_header;
1186 6: optional IndexPageHeader index_page_header;
1187 7: optional DictionaryPageHeader dictionary_page_header;
1188 8: optional DataPageHeaderV2 data_page_header_v2;
1189}
1190);
1191
1192impl PageHeader {
1193 pub(crate) fn read_thrift_without_stats<'a, R>(prot: &mut R) -> Result<Self>
1197 where
1198 R: ThriftCompactInputProtocol<'a>,
1199 {
1200 let mut type_: Option<PageType> = None;
1201 let mut uncompressed_page_size: Option<i32> = None;
1202 let mut compressed_page_size: Option<i32> = None;
1203 let mut crc: Option<i32> = None;
1204 let mut data_page_header: Option<DataPageHeader> = None;
1205 let mut index_page_header: Option<IndexPageHeader> = None;
1206 let mut dictionary_page_header: Option<DictionaryPageHeader> = None;
1207 let mut data_page_header_v2: Option<DataPageHeaderV2> = None;
1208 let mut last_field_id = 0i16;
1209 loop {
1210 let field_ident = prot.read_field_begin(last_field_id)?;
1211 if field_ident.field_type == FieldType::Stop {
1212 break;
1213 }
1214 match field_ident.id {
1215 1 => {
1216 let val = PageType::read_thrift(&mut *prot)?;
1217 type_ = Some(val);
1218 }
1219 2 => {
1220 let val = i32::read_thrift(&mut *prot)?;
1221 uncompressed_page_size = Some(val);
1222 }
1223 3 => {
1224 let val = i32::read_thrift(&mut *prot)?;
1225 compressed_page_size = Some(val);
1226 }
1227 4 => {
1228 let val = i32::read_thrift(&mut *prot)?;
1229 crc = Some(val);
1230 }
1231 5 => {
1232 let val = DataPageHeader::read_thrift_without_stats(&mut *prot)?;
1233 data_page_header = Some(val);
1234 }
1235 6 => {
1236 let val = IndexPageHeader::read_thrift(&mut *prot)?;
1237 index_page_header = Some(val);
1238 }
1239 7 => {
1240 let val = DictionaryPageHeader::read_thrift(&mut *prot)?;
1241 dictionary_page_header = Some(val);
1242 }
1243 8 => {
1244 let val = DataPageHeaderV2::read_thrift_without_stats(&mut *prot)?;
1245 data_page_header_v2 = Some(val);
1246 }
1247 _ => {
1248 prot.skip(field_ident.field_type)?;
1249 }
1250 }
1251 last_field_id = field_ident.id;
1252 }
1253 let Some(type_) = type_ else {
1254 return Err(general_err!("Required field type_ is missing"));
1255 };
1256 let Some(uncompressed_page_size) = uncompressed_page_size else {
1257 return Err(general_err!(
1258 "Required field uncompressed_page_size is missing"
1259 ));
1260 };
1261 let Some(compressed_page_size) = compressed_page_size else {
1262 return Err(general_err!(
1263 "Required field compressed_page_size is missing"
1264 ));
1265 };
1266 Ok(Self {
1267 r#type: type_,
1268 uncompressed_page_size,
1269 compressed_page_size,
1270 crc,
1271 data_page_header,
1272 index_page_header,
1273 dictionary_page_header,
1274 data_page_header_v2,
1275 })
1276 }
1277}
1278
1279#[cfg(feature = "encryption")]
1283fn should_write_column_stats(column_chunk: &ColumnChunkMetaData) -> bool {
1284 column_chunk.encrypted_column_metadata.is_none()
1288}
1289
1290#[cfg(not(feature = "encryption"))]
1291fn should_write_column_stats(_column_chunk: &ColumnChunkMetaData) -> bool {
1292 true
1293}
1294
1295pub(super) fn serialize_column_meta_data<W: Write>(
1316 column_chunk: &ColumnChunkMetaData,
1317 w: &mut ThriftCompactOutputProtocol<W>,
1318) -> Result<()> {
1319 use crate::file::statistics::page_stats_to_thrift;
1320
1321 column_chunk.column_type().write_thrift_field(w, 1, 0)?;
1322 column_chunk
1323 .encodings()
1324 .collect::<Vec<_>>()
1325 .write_thrift_field(w, 2, 1)?;
1326 if w.write_path_in_schema() {
1327 let path = column_chunk.column_descr.path().parts();
1328 let path: Vec<&str> = path.iter().map(|v| v.as_str()).collect();
1329 path.write_thrift_field(w, 3, 2)?;
1330 column_chunk.compression.write_thrift_field(w, 4, 3)?;
1331 } else {
1332 column_chunk.compression.write_thrift_field(w, 4, 2)?;
1333 }
1334
1335 column_chunk.num_values.write_thrift_field(w, 5, 4)?;
1336 column_chunk
1337 .total_uncompressed_size
1338 .write_thrift_field(w, 6, 5)?;
1339 column_chunk
1340 .total_compressed_size
1341 .write_thrift_field(w, 7, 6)?;
1342 let mut last_field_id = column_chunk.data_page_offset.write_thrift_field(w, 9, 7)?;
1344 if let Some(index_page_offset) = column_chunk.index_page_offset {
1345 last_field_id = index_page_offset.write_thrift_field(w, 10, last_field_id)?;
1346 }
1347 if let Some(dictionary_page_offset) = column_chunk.dictionary_page_offset {
1348 last_field_id = dictionary_page_offset.write_thrift_field(w, 11, last_field_id)?;
1349 }
1350
1351 if should_write_column_stats(column_chunk) {
1352 let stats = page_stats_to_thrift(column_chunk.statistics());
1354 if let Some(stats) = stats {
1355 last_field_id = stats.write_thrift_field(w, 12, last_field_id)?;
1356 }
1357 if let Some(page_encoding_stats) = column_chunk.page_encoding_stats() {
1358 last_field_id = page_encoding_stats.write_thrift_field(w, 13, last_field_id)?;
1359 }
1360 if let Some(bloom_filter_offset) = column_chunk.bloom_filter_offset {
1361 last_field_id = bloom_filter_offset.write_thrift_field(w, 14, last_field_id)?;
1362 }
1363 if let Some(bloom_filter_length) = column_chunk.bloom_filter_length {
1364 last_field_id = bloom_filter_length.write_thrift_field(w, 15, last_field_id)?;
1365 }
1366
1367 let size_stats = if column_chunk.unencoded_byte_array_data_bytes.is_some()
1369 || column_chunk.repetition_level_histogram.is_some()
1370 || column_chunk.definition_level_histogram.is_some()
1371 {
1372 let repetition_level_histogram = column_chunk
1373 .repetition_level_histogram()
1374 .map(|hist| hist.clone().into_inner());
1375
1376 let definition_level_histogram = column_chunk
1377 .definition_level_histogram()
1378 .map(|hist| hist.clone().into_inner());
1379
1380 Some(SizeStatistics {
1381 unencoded_byte_array_data_bytes: column_chunk.unencoded_byte_array_data_bytes,
1382 repetition_level_histogram,
1383 definition_level_histogram,
1384 })
1385 } else {
1386 None
1387 };
1388 if let Some(size_stats) = size_stats {
1389 last_field_id = size_stats.write_thrift_field(w, 16, last_field_id)?;
1390 }
1391
1392 if let Some(geo_stats) = column_chunk.geo_statistics() {
1393 geo_stats.write_thrift_field(w, 17, last_field_id)?;
1394 }
1395 }
1396
1397 w.write_struct_end()
1398}
1399
1400pub(super) struct FileMeta<'a> {
1402 pub(super) file_metadata: &'a crate::file::metadata::FileMetaData,
1403 pub(super) row_groups: &'a Vec<RowGroupMetaData>,
1404 pub(super) write_path_in_schema: bool,
1406}
1407
1408impl WriteThrift for FileMeta<'_> {
1420 const ELEMENT_TYPE: ElementType = ElementType::Struct;
1421
1422 #[cfg_attr(not(feature = "encryption"), expect(unused_assignments))]
1424 fn write_thrift<W: Write>(&self, writer: &mut ThriftCompactOutputProtocol<W>) -> Result<()> {
1425 writer.set_write_path_in_schema(self.write_path_in_schema);
1426 writer.set_write_row_group_ordinal(i16::try_from(self.row_groups.len()).is_ok());
1428
1429 self.file_metadata
1430 .version
1431 .write_thrift_field(writer, 1, 0)?;
1432
1433 let root = self.file_metadata.schema_descr().root_schema_ptr();
1436 let schema_len = num_nodes(&root)?;
1437 writer.write_field_begin(FieldType::List, 2, 1)?;
1438 writer.write_list_begin(ElementType::Struct, schema_len)?;
1439 write_schema(&root, writer)?;
1441
1442 self.file_metadata
1443 .num_rows
1444 .write_thrift_field(writer, 3, 2)?;
1445
1446 let mut last_field_id = self.row_groups.write_thrift_field(writer, 4, 3)?;
1448
1449 if let Some(kv_metadata) = self.file_metadata.key_value_metadata() {
1450 last_field_id = kv_metadata.write_thrift_field(writer, 5, last_field_id)?;
1451 }
1452 if let Some(created_by) = self.file_metadata.created_by() {
1453 last_field_id = created_by.write_thrift_field(writer, 6, last_field_id)?;
1454 }
1455 if let Some(column_orders) = self.file_metadata.column_orders() {
1456 last_field_id = column_orders.write_thrift_field(writer, 7, last_field_id)?;
1457 }
1458 #[cfg(feature = "encryption")]
1459 if let Some(algo) = self.file_metadata.encryption_algorithm.as_ref() {
1460 last_field_id = algo.write_thrift_field(writer, 8, last_field_id)?;
1461 }
1462 #[cfg(feature = "encryption")]
1463 if let Some(key) = self.file_metadata.footer_signing_key_metadata.as_ref() {
1464 key.as_slice()
1465 .write_thrift_field(writer, 9, last_field_id)?;
1466 }
1467
1468 writer.write_struct_end()
1469 }
1470}
1471
1472fn write_schema<W: Write>(
1473 schema: &TypePtr,
1474 writer: &mut ThriftCompactOutputProtocol<W>,
1475) -> Result<()> {
1476 if !schema.is_group() {
1477 return Err(general_err!("Root schema must be Group type"));
1478 }
1479 write_schema_helper(schema, writer)
1480}
1481
1482fn write_schema_helper<W: Write>(
1483 node: &TypePtr,
1484 writer: &mut ThriftCompactOutputProtocol<W>,
1485) -> Result<()> {
1486 match node.as_ref() {
1487 crate::schema::types::Type::PrimitiveType {
1488 basic_info,
1489 physical_type,
1490 type_length,
1491 scale,
1492 precision,
1493 } => {
1494 let element = SchemaElement {
1495 r#type: Some(*physical_type),
1496 type_length: if *type_length >= 0 {
1497 Some(*type_length)
1498 } else {
1499 None
1500 },
1501 repetition_type: Some(basic_info.repetition()),
1502 name: basic_info.name(),
1503 num_children: None,
1504 converted_type: match basic_info.converted_type() {
1505 ConvertedType::NONE => None,
1506 other => Some(other),
1507 },
1508 scale: if *scale >= 0 { Some(*scale) } else { None },
1509 precision: if *precision >= 0 {
1510 Some(*precision)
1511 } else {
1512 None
1513 },
1514 field_id: if basic_info.has_id() {
1515 Some(basic_info.id())
1516 } else {
1517 None
1518 },
1519 logical_type: basic_info.logical_type_ref().cloned(),
1520 };
1521 element.write_thrift(writer)
1522 }
1523 crate::schema::types::Type::GroupType { basic_info, fields } => {
1524 let repetition = if basic_info.has_repetition() {
1525 Some(basic_info.repetition())
1526 } else {
1527 None
1528 };
1529
1530 let element = SchemaElement {
1531 r#type: None,
1532 type_length: None,
1533 repetition_type: repetition,
1534 name: basic_info.name(),
1535 num_children: Some(fields.len().try_into()?),
1536 converted_type: match basic_info.converted_type() {
1537 ConvertedType::NONE => None,
1538 other => Some(other),
1539 },
1540 scale: None,
1541 precision: None,
1542 field_id: if basic_info.has_id() {
1543 Some(basic_info.id())
1544 } else {
1545 None
1546 },
1547 logical_type: basic_info.logical_type_ref().cloned(),
1548 };
1549
1550 element.write_thrift(writer)?;
1551
1552 for field in fields {
1554 write_schema_helper(field, writer)?;
1555 }
1556 Ok(())
1557 }
1558 }
1559}
1560
1561impl WriteThrift for RowGroupMetaData {
1571 const ELEMENT_TYPE: ElementType = ElementType::Struct;
1572
1573 fn write_thrift<W: Write>(&self, writer: &mut ThriftCompactOutputProtocol<W>) -> Result<()> {
1574 self.columns.write_thrift_field(writer, 1, 0)?;
1576 self.total_byte_size.write_thrift_field(writer, 2, 1)?;
1577 let mut last_field_id = self.num_rows.write_thrift_field(writer, 3, 2)?;
1578 if let Some(sorting_columns) = self.sorting_columns() {
1579 last_field_id = sorting_columns.write_thrift_field(writer, 4, last_field_id)?;
1580 }
1581 if let Some(file_offset) = self.file_offset() {
1582 last_field_id = file_offset.write_thrift_field(writer, 5, last_field_id)?;
1583 }
1584 last_field_id = self
1586 .compressed_size()
1587 .write_thrift_field(writer, 6, last_field_id)?;
1588
1589 if writer.write_row_group_ordinal()
1591 && let Some(ordinal) = self.ordinal()
1592 && let Ok(ordinal) = i16::try_from(ordinal)
1593 {
1594 ordinal.write_thrift_field(writer, 7, last_field_id)?;
1595 }
1596 writer.write_struct_end()
1597 }
1598}
1599
1600impl WriteThrift for ColumnChunkMetaData {
1612 const ELEMENT_TYPE: ElementType = ElementType::Struct;
1613
1614 #[cfg_attr(not(feature = "encryption"), expect(unused_assignments))]
1615 fn write_thrift<W: Write>(&self, writer: &mut ThriftCompactOutputProtocol<W>) -> Result<()> {
1616 let mut last_field_id = 0i16;
1617 if let Some(file_path) = self.file_path() {
1618 last_field_id = file_path.write_thrift_field(writer, 1, last_field_id)?;
1619 }
1620 last_field_id = self
1621 .file_offset()
1622 .write_thrift_field(writer, 2, last_field_id)?;
1623
1624 #[cfg(feature = "encryption")]
1625 let write_meta_data =
1626 self.encrypted_column_metadata.is_none() || self.plaintext_footer_mode;
1627 #[cfg(not(feature = "encryption"))]
1628 let write_meta_data = true;
1629
1630 if write_meta_data {
1636 writer.write_field_begin(FieldType::Struct, 3, last_field_id)?;
1637 serialize_column_meta_data(self, writer)?;
1638 last_field_id = 3;
1639 }
1640
1641 if let Some(offset_idx_off) = self.offset_index_offset() {
1642 last_field_id = offset_idx_off.write_thrift_field(writer, 4, last_field_id)?;
1643 }
1644 if let Some(offset_idx_len) = self.offset_index_length() {
1645 last_field_id = offset_idx_len.write_thrift_field(writer, 5, last_field_id)?;
1646 }
1647 if let Some(column_idx_off) = self.column_index_offset() {
1648 last_field_id = column_idx_off.write_thrift_field(writer, 6, last_field_id)?;
1649 }
1650 if let Some(column_idx_len) = self.column_index_length() {
1651 last_field_id = column_idx_len.write_thrift_field(writer, 7, last_field_id)?;
1652 }
1653 #[cfg(feature = "encryption")]
1654 {
1655 if let Some(crypto_metadata) = self.crypto_metadata() {
1656 last_field_id = crypto_metadata.write_thrift_field(writer, 8, last_field_id)?;
1657 }
1658 if let Some(encrypted_meta) = self.encrypted_column_metadata.as_ref() {
1659 encrypted_meta
1660 .as_slice()
1661 .write_thrift_field(writer, 9, last_field_id)?;
1662 }
1663 }
1664
1665 writer.write_struct_end()
1666 }
1667}
1668
1669impl WriteThrift for crate::geospatial::statistics::GeospatialStatistics {
1674 const ELEMENT_TYPE: ElementType = ElementType::Struct;
1675
1676 fn write_thrift<W: Write>(&self, writer: &mut ThriftCompactOutputProtocol<W>) -> Result<()> {
1677 let mut last_field_id = 0i16;
1678 if let Some(bbox) = self.bounding_box() {
1679 last_field_id = bbox.write_thrift_field(writer, 1, last_field_id)?;
1680 }
1681 if let Some(geo_types) = self.geospatial_types() {
1682 geo_types.write_thrift_field(writer, 2, last_field_id)?;
1683 }
1684
1685 writer.write_struct_end()
1686 }
1687}
1688
1689use crate::geospatial::statistics::GeospatialStatistics as RustGeospatialStatistics;
1691write_thrift_field!(RustGeospatialStatistics, FieldType::Struct);
1692
1693impl WriteThrift for crate::geospatial::bounding_box::BoundingBox {
1704 const ELEMENT_TYPE: ElementType = ElementType::Struct;
1705
1706 fn write_thrift<W: Write>(&self, writer: &mut ThriftCompactOutputProtocol<W>) -> Result<()> {
1707 self.get_xmin().write_thrift_field(writer, 1, 0)?;
1708 self.get_xmax().write_thrift_field(writer, 2, 1)?;
1709 self.get_ymin().write_thrift_field(writer, 3, 2)?;
1710 let mut last_field_id = self.get_ymax().write_thrift_field(writer, 4, 3)?;
1711
1712 if let Some(zmin) = self.get_zmin() {
1713 last_field_id = zmin.write_thrift_field(writer, 5, last_field_id)?;
1714 }
1715 if let Some(zmax) = self.get_zmax() {
1716 last_field_id = zmax.write_thrift_field(writer, 6, last_field_id)?;
1717 }
1718 if let Some(mmin) = self.get_mmin() {
1719 last_field_id = mmin.write_thrift_field(writer, 7, last_field_id)?;
1720 }
1721 if let Some(mmax) = self.get_mmax() {
1722 mmax.write_thrift_field(writer, 8, last_field_id)?;
1723 }
1724
1725 writer.write_struct_end()
1726 }
1727}
1728
1729use crate::geospatial::bounding_box::BoundingBox as RustBoundingBox;
1731write_thrift_field!(RustBoundingBox, FieldType::Struct);
1732
1733#[cfg(test)]
1734pub(crate) mod tests {
1735 use crate::basic::{Encoding, PageType, Type as PhysicalType};
1736 use crate::errors::Result;
1737 use crate::file::metadata::thrift::{
1738 BoundingBox, DataPageHeaderV2, DictionaryPageHeader, PageHeader, SchemaElement,
1739 write_schema,
1740 };
1741 use crate::file::metadata::{ColumnChunkMetaData, ParquetMetaDataOptions, RowGroupMetaData};
1742 use crate::parquet_thrift::tests::test_roundtrip;
1743 use crate::parquet_thrift::{
1744 ElementType, ThriftCompactOutputProtocol, ThriftSliceInputProtocol, WriteThrift,
1745 read_thrift_vec,
1746 };
1747 use crate::schema::types::{
1748 ColumnDescriptor, ColumnPath, SchemaDescriptor, TypePtr, num_nodes,
1749 parquet_schema_from_array,
1750 };
1751 use std::sync::Arc;
1752
1753 pub(crate) fn read_row_group(
1755 buf: &[u8],
1756 schema_descr: Arc<SchemaDescriptor>,
1757 ) -> Result<RowGroupMetaData> {
1758 let mut reader = ThriftSliceInputProtocol::new(buf);
1759 crate::file::metadata::thrift::read_row_group(&mut reader, &schema_descr, None)
1760 }
1761
1762 pub(crate) fn read_column_chunk(
1763 buf: &[u8],
1764 column_descr: Arc<ColumnDescriptor>,
1765 ) -> Result<ColumnChunkMetaData> {
1766 read_column_chunk_with_options(buf, column_descr, None)
1767 }
1768
1769 pub(crate) fn read_column_chunk_with_options(
1770 buf: &[u8],
1771 column_descr: Arc<ColumnDescriptor>,
1772 options: Option<&ParquetMetaDataOptions>,
1773 ) -> Result<ColumnChunkMetaData> {
1774 let mut reader = ThriftSliceInputProtocol::new(buf);
1775 crate::file::metadata::thrift::read_column_chunk(&mut reader, &column_descr, 0, options)
1776 }
1777
1778 pub(crate) fn roundtrip_schema(schema: TypePtr) -> Result<TypePtr> {
1779 let num_nodes = num_nodes(&schema)?;
1780 let mut buf = Vec::new();
1781 let mut writer = ThriftCompactOutputProtocol::new(&mut buf);
1782
1783 writer.write_list_begin(ElementType::Struct, num_nodes)?;
1785
1786 write_schema(&schema, &mut writer)?;
1788
1789 let mut prot = ThriftSliceInputProtocol::new(&buf);
1790 let se: Vec<SchemaElement> = read_thrift_vec(&mut prot)?;
1791 parquet_schema_from_array(se)
1792 }
1793
1794 pub(crate) fn schema_to_buf(schema: &TypePtr) -> Result<Vec<u8>> {
1795 let num_nodes = num_nodes(schema)?;
1796 let mut buf = Vec::new();
1797 let mut writer = ThriftCompactOutputProtocol::new(&mut buf);
1798
1799 writer.write_list_begin(ElementType::Struct, num_nodes)?;
1801
1802 write_schema(schema, &mut writer)?;
1804 Ok(buf)
1805 }
1806
1807 pub(crate) fn buf_to_schema_list(buf: &mut Vec<u8>) -> Result<Vec<SchemaElement<'_>>> {
1808 let mut prot = ThriftSliceInputProtocol::new(buf.as_mut_slice());
1809 read_thrift_vec(&mut prot)
1810 }
1811
1812 fn thrift_bytes<T: WriteThrift>(value: &T) -> Vec<u8> {
1813 let mut buf = Vec::new();
1814 let mut writer = ThriftCompactOutputProtocol::new(&mut buf);
1815 value.write_thrift(&mut writer).unwrap();
1816 buf
1817 }
1818
1819 fn change_false_bool_field_to_i32(buf: &mut [u8]) {
1820 let pos = buf
1821 .iter()
1822 .rposition(|byte| *byte == 0x12)
1823 .expect("expected BOOL_FALSE field header byte");
1824 buf[pos] = 0x15;
1825 }
1826
1827 fn assert_malformed_bool_error(err: crate::errors::ParquetError) {
1828 let msg = err.to_string();
1829 assert!(
1830 msg.contains("Unexpected struct field type"),
1831 "unexpected error message: {msg}"
1832 );
1833 }
1834
1835 #[test]
1836 fn test_bounding_box_roundtrip() {
1837 test_roundtrip(BoundingBox {
1838 xmin: 0.1.into(),
1839 xmax: 10.3.into(),
1840 ymin: 0.001.into(),
1841 ymax: 128.5.into(),
1842 zmin: None,
1843 zmax: None,
1844 mmin: None,
1845 mmax: None,
1846 });
1847
1848 test_roundtrip(BoundingBox {
1849 xmin: 0.1.into(),
1850 xmax: 10.3.into(),
1851 ymin: 0.001.into(),
1852 ymax: 128.5.into(),
1853 zmin: Some(11.0.into()),
1854 zmax: Some(1300.0.into()),
1855 mmin: None,
1856 mmax: None,
1857 });
1858
1859 test_roundtrip(BoundingBox {
1860 xmin: 0.1.into(),
1861 xmax: 10.3.into(),
1862 ymin: 0.001.into(),
1863 ymax: 128.5.into(),
1864 zmin: Some(11.0.into()),
1865 zmax: Some(1300.0.into()),
1866 mmin: Some(3.7.into()),
1867 mmax: Some(42.0.into()),
1868 });
1869 }
1870
1871 #[test]
1872 fn test_convert_stats_preserves_missing_null_count() {
1873 let primitive =
1874 crate::schema::types::Type::primitive_type_builder("col", PhysicalType::INT32)
1875 .build()
1876 .unwrap();
1877 let column_descr = Arc::new(ColumnDescriptor::new(
1878 Arc::new(primitive),
1879 0,
1880 0,
1881 ColumnPath::new(vec![]),
1882 ));
1883
1884 let none_null_count = super::Statistics {
1885 max: None,
1886 min: None,
1887 null_count: None,
1888 distinct_count: None,
1889 max_value: None,
1890 min_value: None,
1891 is_max_value_exact: None,
1892 is_min_value_exact: None,
1893 nan_count: None,
1894 };
1895 let decoded_none = super::convert_stats(&column_descr, Some(none_null_count))
1896 .unwrap()
1897 .unwrap();
1898 assert_eq!(decoded_none.null_count_opt(), None);
1899
1900 let zero_null_count = super::Statistics {
1901 max: None,
1902 min: None,
1903 null_count: Some(0),
1904 distinct_count: None,
1905 max_value: None,
1906 min_value: None,
1907 is_max_value_exact: None,
1908 is_min_value_exact: None,
1909 nan_count: None,
1910 };
1911 let decoded_zero = super::convert_stats(&column_descr, Some(zero_null_count))
1912 .unwrap()
1913 .unwrap();
1914 assert_eq!(decoded_zero.null_count_opt(), Some(0));
1915 }
1916
1917 #[test]
1918 fn test_convert_stats_returns_error_for_overlong_int96_statistics() {
1919 let primitive =
1920 crate::schema::types::Type::primitive_type_builder("col", PhysicalType::INT96)
1921 .build()
1922 .unwrap();
1923 let column_descr = Arc::new(ColumnDescriptor::new(
1924 Arc::new(primitive),
1925 0,
1926 0,
1927 ColumnPath::new(vec![]),
1928 ));
1929 let invalid = (0..13).collect::<Vec<_>>();
1930
1931 let make_stats = |min, max| super::Statistics {
1932 max,
1933 min,
1934 null_count: Some(0),
1935 distinct_count: None,
1936 max_value: None,
1937 min_value: None,
1938 is_max_value_exact: None,
1939 is_min_value_exact: None,
1940 nan_count: None,
1941 };
1942
1943 let err = super::convert_stats(&column_descr, Some(make_stats(Some(&invalid), None)))
1944 .unwrap_err();
1945 assert_eq!(
1946 err.to_string(),
1947 "Parquet error: Incorrect Int96 min statistics"
1948 );
1949
1950 let err = super::convert_stats(&column_descr, Some(make_stats(None, Some(&invalid))))
1951 .unwrap_err();
1952 assert_eq!(
1953 err.to_string(),
1954 "Parquet error: Incorrect Int96 max statistics"
1955 );
1956 }
1957
1958 #[test]
1959 fn malformed_bool_field_returns_error_not_panic() {
1960 let page_header = PageHeader {
1961 r#type: PageType::DICTIONARY_PAGE,
1962 uncompressed_page_size: 1,
1963 compressed_page_size: 1,
1964 crc: None,
1965 data_page_header: None,
1966 index_page_header: None,
1967 dictionary_page_header: Some(DictionaryPageHeader {
1968 num_values: 1,
1969 encoding: Encoding::PLAIN,
1970 is_sorted: Some(false),
1971 }),
1972 data_page_header_v2: None,
1973 };
1974
1975 let mut buf = thrift_bytes(&page_header);
1976 change_false_bool_field_to_i32(&mut buf);
1977
1978 let mut prot = ThriftSliceInputProtocol::new(&buf);
1979 let err = PageHeader::read_thrift_without_stats(&mut prot)
1980 .expect_err("malformed bool field should return an error");
1981 assert_malformed_bool_error(err);
1982 }
1983
1984 #[test]
1985 fn malformed_data_page_v2_bool_field_returns_error_not_panic() {
1986 let data_page_header_v2 = DataPageHeaderV2 {
1987 num_values: 1,
1988 num_nulls: 0,
1989 num_rows: 1,
1990 encoding: Encoding::PLAIN,
1991 definition_levels_byte_length: 0,
1992 repetition_levels_byte_length: 0,
1993 is_compressed: Some(false),
1994 statistics: None,
1995 };
1996
1997 let mut buf = thrift_bytes(&data_page_header_v2);
1998 change_false_bool_field_to_i32(&mut buf);
1999
2000 let mut prot = ThriftSliceInputProtocol::new(&buf);
2001 let err = DataPageHeaderV2::read_thrift_without_stats(&mut prot)
2002 .expect_err("malformed bool field should return an error");
2003 assert_malformed_bool_error(err);
2004 }
2005
2006 fn roundtrip_rg_ordinals(ordinals: &[Option<i32>]) -> Vec<Option<i32>> {
2010 use crate::file::metadata::ParquetMetaDataWriter;
2011 use crate::file::metadata::{FileMetaData, ParquetMetaData, ParquetMetaDataReader};
2012 use crate::schema::types::Type as SchemaType;
2013
2014 let field = SchemaType::primitive_type_builder("c", PhysicalType::INT32)
2015 .build()
2016 .unwrap();
2017 let schema = SchemaType::group_type_builder("schema")
2018 .with_fields(vec![Arc::new(field)])
2019 .build()
2020 .unwrap();
2021 let schema_descr = Arc::new(SchemaDescriptor::new(Arc::new(schema)));
2022
2023 let row_groups = ordinals
2024 .iter()
2025 .map(|ordinal| {
2026 let columns = schema_descr
2027 .columns()
2028 .iter()
2029 .map(|col| ColumnChunkMetaData::builder(col.clone()).build().unwrap())
2030 .collect();
2031 let mut builder =
2032 crate::file::metadata::RowGroupMetaData::builder(schema_descr.clone())
2033 .set_num_rows(10)
2034 .set_total_byte_size(100)
2035 .set_column_metadata(columns);
2036 if let Some(ordinal) = ordinal {
2037 builder = builder.set_ordinal(*ordinal);
2038 }
2039 builder.build().unwrap()
2040 })
2041 .collect();
2042
2043 let file_metadata = FileMetaData::new(
2044 1,
2045 10 * ordinals.len() as i64,
2046 None,
2047 None,
2048 schema_descr,
2049 None,
2050 );
2051 let metadata = ParquetMetaData::new(file_metadata, row_groups);
2052
2053 let mut buffer = Vec::new();
2054 ParquetMetaDataWriter::new(&mut buffer, &metadata)
2055 .finish()
2056 .unwrap();
2057 let decoded = ParquetMetaDataReader::decode_metadata(&buffer[..buffer.len() - 8]).unwrap();
2059 decoded.row_groups().iter().map(|rg| rg.ordinal()).collect()
2060 }
2061
2062 #[test]
2065 fn ordinals_all_present_are_honored() {
2066 assert_eq!(
2067 roundtrip_rg_ordinals(&[Some(5), Some(1), Some(3)]),
2068 vec![Some(5), Some(1), Some(3)],
2069 );
2070 }
2071
2072 #[test]
2075 fn ordinals_none_present_are_sequentially_filled() {
2076 assert_eq!(
2077 roundtrip_rg_ordinals(&[None, None, None]),
2078 vec![Some(0), Some(1), Some(2)],
2079 );
2080 }
2081
2082 #[test]
2087 fn ordinals_mixed_decode_succeeds_untouched() {
2088 assert_eq!(
2089 roundtrip_rg_ordinals(&[Some(0), None, Some(2)]),
2090 vec![Some(0), None, Some(2)],
2091 );
2092 assert_eq!(
2094 roundtrip_rg_ordinals(&[None, Some(1), Some(2)]),
2095 vec![None, Some(1), Some(2)],
2096 );
2097 }
2098}