1use base64::Engine;
21use base64::prelude::BASE64_STANDARD;
22use std::collections::HashMap;
23use std::sync::Arc;
24
25use arrow_ipc::writer;
26use arrow_schema::{DataType, Field, FieldRef, Fields, Schema, TimeUnit};
27
28use crate::basic::{
29 ConvertedType, LogicalType, Repetition, TimeUnit as ParquetTimeUnit, Type as PhysicalType,
30};
31use crate::errors::{ParquetError, Result};
32use crate::file::{metadata::KeyValue, properties::WriterProperties};
33use crate::schema::types::{ColumnDescriptor, SchemaDescriptor, Type};
34
35mod complex;
36mod extension;
37mod primitive;
38pub mod virtual_type;
39
40use super::PARQUET_FIELD_ID_META_KEY;
41use crate::arrow::ProjectionMask;
42use crate::arrow::schema::extension::{
43 has_extension_type, logical_type_for_binary, logical_type_for_binary_view,
44 logical_type_for_fixed_size_binary, logical_type_for_string, logical_type_for_struct,
45 try_add_extension_type,
46};
47pub(crate) use complex::{ParquetField, ParquetFieldType, VirtualColumnType};
48
49pub fn parquet_to_arrow_schema(
54 parquet_schema: &SchemaDescriptor,
55 key_value_metadata: Option<&Vec<KeyValue>>,
56) -> Result<Schema> {
57 parquet_to_arrow_schema_by_columns(parquet_schema, ProjectionMask::all(), key_value_metadata)
58}
59
60pub fn parquet_to_arrow_schema_by_columns(
63 parquet_schema: &SchemaDescriptor,
64 mask: ProjectionMask,
65 key_value_metadata: Option<&Vec<KeyValue>>,
66) -> Result<Schema> {
67 Ok(parquet_to_arrow_schema_and_fields(parquet_schema, mask, key_value_metadata, &[])?.0)
68}
69
70pub(crate) fn parquet_to_arrow_schema_and_fields(
76 parquet_schema: &SchemaDescriptor,
77 mask: ProjectionMask,
78 key_value_metadata: Option<&Vec<KeyValue>>,
79 virtual_columns: &[FieldRef],
80) -> Result<(Schema, Option<ParquetField>)> {
81 let mut metadata = parse_key_value_metadata(key_value_metadata).unwrap_or_default();
82 let maybe_schema = metadata
83 .remove(super::ARROW_SCHEMA_META_KEY)
84 .map(|value| get_arrow_schema_from_metadata(&value))
85 .transpose()?;
86
87 if let Some(arrow_schema) = &maybe_schema {
89 arrow_schema.metadata().iter().for_each(|(k, v)| {
90 metadata.entry(k.clone()).or_insert_with(|| v.clone());
91 });
92 }
93
94 let hint = maybe_schema.as_ref().map(|s| s.fields());
95 let field_levels =
96 parquet_to_arrow_field_levels_with_virtual(parquet_schema, mask, hint, virtual_columns)?;
97 let schema = Schema::new_with_metadata(field_levels.fields, metadata);
98 Ok((schema, field_levels.levels))
99}
100
101#[derive(Debug, Clone)]
109pub struct FieldLevels {
110 pub(crate) fields: Fields,
111 pub(crate) levels: Option<ParquetField>,
112}
113
114pub fn parquet_to_arrow_field_levels(
135 schema: &SchemaDescriptor,
136 mask: ProjectionMask,
137 hint: Option<&Fields>,
138) -> Result<FieldLevels> {
139 parquet_to_arrow_field_levels_with_virtual(schema, mask, hint, &[])
140}
141
142pub fn parquet_to_arrow_field_levels_with_virtual(
172 schema: &SchemaDescriptor,
173 mask: ProjectionMask,
174 hint: Option<&Fields>,
175 virtual_columns: &[FieldRef],
176) -> Result<FieldLevels> {
177 for field in virtual_columns {
179 if !virtual_type::is_virtual_column(field) {
180 return Err(ParquetError::General(format!(
181 "Field '{}' is not a virtual column. Virtual columns must have extension type names starting with 'arrow.virtual.'",
182 field.name()
183 )));
184 }
185 }
186
187 let mut parquet_field = match complex::convert_schema(schema, mask, hint)? {
189 Some(field) => field,
190 None if virtual_columns.is_empty() => {
191 return Ok(FieldLevels {
192 fields: Fields::empty(),
193 levels: None,
194 });
195 }
196 None => {
197 ParquetField {
199 rep_level: 0,
200 def_level: 0,
201 nullable: false,
202 arrow_type: DataType::Struct(Fields::empty()),
203 field_type: ParquetFieldType::Group {
204 children: Vec::new(),
205 },
206 }
207 }
208 };
209
210 if !virtual_columns.is_empty() {
212 match &mut parquet_field.field_type {
213 ParquetFieldType::Group { children } => {
214 let DataType::Struct(ref mut fields) = parquet_field.arrow_type else {
216 unreachable!("Root field must be a struct");
217 };
218
219 let mut fields_vec: Vec<FieldRef> = fields.iter().cloned().collect();
221
222 for virtual_column in virtual_columns {
224 assert_eq!(
226 parquet_field.rep_level, 0,
227 "Virtual columns can only be added at rep level 0"
228 );
229 assert_eq!(
230 parquet_field.def_level, 0,
231 "Virtual columns can only be added at def level 0"
232 );
233
234 fields_vec.push(virtual_column.clone());
235 let virtual_parquet_field = complex::convert_virtual_field(
236 virtual_column,
237 parquet_field.rep_level,
238 parquet_field.def_level,
239 )?;
240 children.push(virtual_parquet_field);
241 }
242
243 parquet_field.arrow_type = DataType::Struct(Fields::from(fields_vec));
245 }
246 _ => unreachable!("Root field must be a group"),
247 }
248 }
249
250 match &parquet_field.arrow_type {
251 DataType::Struct(fields) => Ok(FieldLevels {
252 fields: fields.clone(),
253 levels: Some(parquet_field),
254 }),
255 _ => unreachable!(),
256 }
257}
258
259fn get_arrow_schema_from_metadata(encoded_meta: &str) -> Result<Schema> {
261 let decoded = BASE64_STANDARD.decode(encoded_meta);
262 match decoded {
263 Ok(bytes) => {
264 let slice = if bytes.len() > 8 && bytes[0..4] == [255u8; 4] {
265 &bytes[8..]
266 } else {
267 bytes.as_slice()
268 };
269 match arrow_ipc::root_as_message(slice) {
270 Ok(message) => {
271 let schema = message
272 .header_as_schema()
273 .ok_or_else(|| arrow_err!("the message is not Arrow Schema"))?;
274 arrow_ipc::convert::try_fb_to_schema(schema).map_err(Into::into)
275 }
276 Err(err) => {
277 Err(arrow_err!(
279 "Unable to get root as message stored in {}: {:?}",
280 super::ARROW_SCHEMA_META_KEY,
281 err
282 ))
283 }
284 }
285 }
286 Err(err) => {
287 Err(arrow_err!(
289 "Unable to decode the encoded schema stored in {}, {:?}",
290 super::ARROW_SCHEMA_META_KEY,
291 err
292 ))
293 }
294 }
295}
296
297pub fn encode_arrow_schema(schema: &Schema) -> String {
299 let options = writer::IpcWriteOptions::default();
300 let mut dictionary_tracker = writer::DictionaryTracker::new(true);
301 let data_gen = writer::IpcDataGenerator::default();
302 let mut serialized_schema =
303 data_gen.schema_to_bytes_with_dictionary_tracker(schema, &mut dictionary_tracker, &options);
304
305 let schema_len = serialized_schema.ipc_message.len();
308 let mut len_prefix_schema = Vec::with_capacity(schema_len + 8);
309 len_prefix_schema.append(&mut vec![255u8, 255, 255, 255]);
310 len_prefix_schema.append((schema_len as u32).to_le_bytes().to_vec().as_mut());
311 len_prefix_schema.append(&mut serialized_schema.ipc_message);
312
313 BASE64_STANDARD.encode(&len_prefix_schema)
314}
315
316fn flatten_ree_field(field: &Field) -> Field {
317 match field.data_type() {
318 DataType::RunEndEncoded(_, value_field) => field
319 .clone()
320 .with_data_type(value_field.data_type().clone()),
321 _ => field.clone(),
322 }
323}
324
325pub fn add_encoded_arrow_schema_to_metadata(schema: &Schema, props: &mut WriterProperties) {
332 let has_ree = schema
333 .fields()
334 .iter()
335 .any(|f| matches!(f.data_type(), DataType::RunEndEncoded(_, _)));
336 let flat_schema;
337 let schema = if has_ree {
338 let flat_fields: Vec<Field> = schema
339 .fields()
340 .iter()
341 .map(|f| flatten_ree_field(f))
342 .collect();
343 flat_schema = Schema::new_with_metadata(flat_fields, schema.metadata().clone());
344 &flat_schema
345 } else {
346 schema
347 };
348 let encoded = encode_arrow_schema(schema);
349
350 let schema_kv = KeyValue {
351 key: super::ARROW_SCHEMA_META_KEY.to_string(),
352 value: Some(encoded),
353 };
354
355 let meta = props
356 .key_value_metadata
357 .get_or_insert_with(Default::default);
358
359 let schema_meta = meta
361 .iter()
362 .enumerate()
363 .find(|(_, kv)| kv.key.as_str() == super::ARROW_SCHEMA_META_KEY);
364 match schema_meta {
365 Some((i, _)) => {
366 meta.remove(i);
367 meta.push(schema_kv);
368 }
369 None => {
370 meta.push(schema_kv);
371 }
372 }
373}
374
375#[derive(Debug)]
420pub struct ArrowSchemaConverter<'a> {
421 schema_root: &'a str,
423 coerce_types: bool,
427}
428
429impl Default for ArrowSchemaConverter<'_> {
430 fn default() -> Self {
431 Self::new()
432 }
433}
434
435impl<'a> ArrowSchemaConverter<'a> {
436 pub fn new() -> Self {
438 Self {
439 schema_root: "arrow_schema",
440 coerce_types: false,
441 }
442 }
443
444 pub fn with_coerce_types(mut self, coerce_types: bool) -> Self {
475 self.coerce_types = coerce_types;
476 self
477 }
478
479 pub fn schema_root(mut self, schema_root: &'a str) -> Self {
481 self.schema_root = schema_root;
482 self
483 }
484
485 pub fn convert(&self, schema: &Schema) -> Result<SchemaDescriptor> {
489 let fields = schema
490 .fields()
491 .iter()
492 .map(|field| arrow_to_parquet_type(field, self.coerce_types).map(Arc::new))
493 .collect::<Result<_>>()?;
494 let group = Type::group_type_builder(self.schema_root)
495 .with_fields(fields)
496 .build()?;
497 Ok(SchemaDescriptor::new(Arc::new(group)))
498 }
499}
500
501fn parse_key_value_metadata(
502 key_value_metadata: Option<&Vec<KeyValue>>,
503) -> Option<HashMap<String, String>> {
504 match key_value_metadata {
505 Some(key_values) => {
506 let map: HashMap<String, String> = key_values
507 .iter()
508 .filter_map(|kv| {
509 kv.value
510 .as_ref()
511 .map(|value| (kv.key.clone(), value.clone()))
512 })
513 .collect();
514
515 if map.is_empty() { None } else { Some(map) }
516 }
517 None => None,
518 }
519}
520
521pub fn parquet_to_arrow_field(parquet_column: &ColumnDescriptor) -> Result<Field> {
523 let field = complex::convert_type(&parquet_column.self_type_ptr())?;
524 let mut ret = Field::new(parquet_column.name(), field.arrow_type, field.nullable);
525
526 let parquet_type = parquet_column.self_type();
527 let basic_info = parquet_type.get_basic_info();
528
529 let mut hash_map_size = 0;
530 if basic_info.has_id() {
531 hash_map_size += 1;
532 }
533 if has_extension_type(parquet_type) {
534 hash_map_size += 1;
535 }
536 if hash_map_size == 0 {
537 return Ok(ret);
538 }
539 ret.set_metadata(HashMap::with_capacity(hash_map_size));
540 if basic_info.has_id() {
541 ret.metadata_mut().insert(
542 PARQUET_FIELD_ID_META_KEY.to_string(),
543 basic_info.id().to_string(),
544 );
545 }
546 try_add_extension_type(ret, parquet_column.self_type())
547}
548
549pub fn decimal_length_from_precision(precision: u8) -> usize {
550 (((10.0_f64.powi(precision as i32) + 1.0).log2() + 1.0) / 8.0).ceil() as usize
558}
559
560fn arrow_to_parquet_type(field: &Field, coerce_types: bool) -> Result<Type> {
562 const PARQUET_LIST_ELEMENT_NAME: &str = "element";
563 const PARQUET_MAP_STRUCT_NAME: &str = "key_value";
564 const PARQUET_KEY_FIELD_NAME: &str = "key";
565 const PARQUET_VALUE_FIELD_NAME: &str = "value";
566
567 let name = field.name().as_str();
568 let repetition = if field.is_nullable() {
569 Repetition::OPTIONAL
570 } else {
571 Repetition::REQUIRED
572 };
573 let id = field_id(field);
574 match field.data_type() {
576 DataType::Null => Type::primitive_type_builder(name, PhysicalType::INT32)
577 .with_logical_type(Some(LogicalType::Unknown))
578 .with_repetition(repetition)
579 .with_id(id)
580 .build(),
581 DataType::Boolean => Type::primitive_type_builder(name, PhysicalType::BOOLEAN)
582 .with_repetition(repetition)
583 .with_id(id)
584 .build(),
585 DataType::Int8 => Type::primitive_type_builder(name, PhysicalType::INT32)
586 .with_logical_type(Some(LogicalType::integer(8, true)))
587 .with_repetition(repetition)
588 .with_id(id)
589 .build(),
590 DataType::Int16 => Type::primitive_type_builder(name, PhysicalType::INT32)
591 .with_logical_type(Some(LogicalType::integer(16, true)))
592 .with_repetition(repetition)
593 .with_id(id)
594 .build(),
595 DataType::Int32 => Type::primitive_type_builder(name, PhysicalType::INT32)
596 .with_repetition(repetition)
597 .with_id(id)
598 .build(),
599 DataType::Int64 => Type::primitive_type_builder(name, PhysicalType::INT64)
600 .with_repetition(repetition)
601 .with_id(id)
602 .build(),
603 DataType::UInt8 => Type::primitive_type_builder(name, PhysicalType::INT32)
604 .with_logical_type(Some(LogicalType::integer(8, false)))
605 .with_repetition(repetition)
606 .with_id(id)
607 .build(),
608 DataType::UInt16 => Type::primitive_type_builder(name, PhysicalType::INT32)
609 .with_logical_type(Some(LogicalType::integer(16, false)))
610 .with_repetition(repetition)
611 .with_id(id)
612 .build(),
613 DataType::UInt32 => Type::primitive_type_builder(name, PhysicalType::INT32)
614 .with_logical_type(Some(LogicalType::integer(32, false)))
615 .with_repetition(repetition)
616 .with_id(id)
617 .build(),
618 DataType::UInt64 => Type::primitive_type_builder(name, PhysicalType::INT64)
619 .with_logical_type(Some(LogicalType::integer(64, false)))
620 .with_repetition(repetition)
621 .with_id(id)
622 .build(),
623 DataType::Float16 => Type::primitive_type_builder(name, PhysicalType::FIXED_LEN_BYTE_ARRAY)
624 .with_repetition(repetition)
625 .with_id(id)
626 .with_logical_type(Some(LogicalType::Float16))
627 .with_length(2)
628 .build(),
629 DataType::Float32 => Type::primitive_type_builder(name, PhysicalType::FLOAT)
630 .with_repetition(repetition)
631 .with_id(id)
632 .build(),
633 DataType::Float64 => Type::primitive_type_builder(name, PhysicalType::DOUBLE)
634 .with_repetition(repetition)
635 .with_id(id)
636 .build(),
637 DataType::Timestamp(TimeUnit::Second, _) => {
638 Type::primitive_type_builder(name, PhysicalType::INT64)
640 .with_repetition(repetition)
641 .with_id(id)
642 .build()
643 }
644 DataType::Timestamp(time_unit, tz) => {
645 Type::primitive_type_builder(name, PhysicalType::INT64)
646 .with_logical_type(Some(LogicalType::timestamp(
647 matches!(tz, Some(z) if !z.as_ref().is_empty()),
649 match time_unit {
650 TimeUnit::Second => unreachable!(),
651 TimeUnit::Millisecond => ParquetTimeUnit::MILLIS,
652 TimeUnit::Microsecond => ParquetTimeUnit::MICROS,
653 TimeUnit::Nanosecond => ParquetTimeUnit::NANOS,
654 },
655 )))
656 .with_repetition(repetition)
657 .with_id(id)
658 .build()
659 }
660 DataType::Date32 => Type::primitive_type_builder(name, PhysicalType::INT32)
661 .with_logical_type(Some(LogicalType::Date))
662 .with_repetition(repetition)
663 .with_id(id)
664 .build(),
665 DataType::Date64 => {
666 if coerce_types {
667 Type::primitive_type_builder(name, PhysicalType::INT32)
668 .with_logical_type(Some(LogicalType::Date))
669 .with_repetition(repetition)
670 .with_id(id)
671 .build()
672 } else {
673 Type::primitive_type_builder(name, PhysicalType::INT64)
674 .with_repetition(repetition)
675 .with_id(id)
676 .build()
677 }
678 }
679 DataType::Time32(TimeUnit::Second) => {
680 Type::primitive_type_builder(name, PhysicalType::INT32)
682 .with_repetition(repetition)
683 .with_id(id)
684 .build()
685 }
686 DataType::Time32(unit) => Type::primitive_type_builder(name, PhysicalType::INT32)
687 .with_logical_type(Some(LogicalType::time(
688 field.metadata().contains_key("adjusted_to_utc"),
689 match unit {
690 TimeUnit::Millisecond => ParquetTimeUnit::MILLIS,
691 u => unreachable!("Invalid unit for Time32: {:?}", u),
692 },
693 )))
694 .with_repetition(repetition)
695 .with_id(id)
696 .build(),
697 DataType::Time64(unit) => Type::primitive_type_builder(name, PhysicalType::INT64)
698 .with_logical_type(Some(LogicalType::time(
699 field.metadata().contains_key("adjusted_to_utc"),
700 match unit {
701 TimeUnit::Microsecond => ParquetTimeUnit::MICROS,
702 TimeUnit::Nanosecond => ParquetTimeUnit::NANOS,
703 u => unreachable!("Invalid unit for Time64: {:?}", u),
704 },
705 )))
706 .with_repetition(repetition)
707 .with_id(id)
708 .build(),
709 DataType::Duration(_) => Type::primitive_type_builder(name, PhysicalType::INT64)
710 .with_repetition(repetition)
711 .with_id(id)
712 .build(),
713 DataType::Interval(_) => {
714 Type::primitive_type_builder(name, PhysicalType::FIXED_LEN_BYTE_ARRAY)
715 .with_converted_type(ConvertedType::INTERVAL)
716 .with_repetition(repetition)
717 .with_id(id)
718 .with_length(12)
719 .build()
720 }
721 DataType::Binary | DataType::LargeBinary => {
722 Type::primitive_type_builder(name, PhysicalType::BYTE_ARRAY)
723 .with_repetition(repetition)
724 .with_id(id)
725 .with_logical_type(logical_type_for_binary(field))
726 .build()
727 }
728 DataType::FixedSizeBinary(length) => {
729 Type::primitive_type_builder(name, PhysicalType::FIXED_LEN_BYTE_ARRAY)
730 .with_repetition(repetition)
731 .with_id(id)
732 .with_length(*length)
733 .with_logical_type(logical_type_for_fixed_size_binary(field))
734 .build()
735 }
736 DataType::BinaryView => Type::primitive_type_builder(name, PhysicalType::BYTE_ARRAY)
737 .with_repetition(repetition)
738 .with_id(id)
739 .with_logical_type(logical_type_for_binary_view(field))
740 .build(),
741 DataType::Decimal32(precision, scale)
742 | DataType::Decimal64(precision, scale)
743 | DataType::Decimal128(precision, scale)
744 | DataType::Decimal256(precision, scale) => {
745 let (physical_type, length) = if *precision > 1 && *precision <= 9 {
748 (PhysicalType::INT32, -1)
749 } else if *precision <= 18 {
750 (PhysicalType::INT64, -1)
751 } else {
752 (
753 PhysicalType::FIXED_LEN_BYTE_ARRAY,
754 decimal_length_from_precision(*precision) as i32,
755 )
756 };
757 Type::primitive_type_builder(name, physical_type)
758 .with_repetition(repetition)
759 .with_id(id)
760 .with_length(length)
761 .with_logical_type(Some(LogicalType::decimal(*scale as i32, *precision as i32)))
762 .with_precision(*precision as i32)
763 .with_scale(*scale as i32)
764 .build()
765 }
766 DataType::Utf8 | DataType::LargeUtf8 => {
767 Type::primitive_type_builder(name, PhysicalType::BYTE_ARRAY)
768 .with_logical_type(logical_type_for_string(field))
769 .with_repetition(repetition)
770 .with_id(id)
771 .build()
772 }
773 DataType::Utf8View => Type::primitive_type_builder(name, PhysicalType::BYTE_ARRAY)
774 .with_logical_type(logical_type_for_string(field))
775 .with_repetition(repetition)
776 .with_id(id)
777 .build(),
778 DataType::List(f)
779 | DataType::FixedSizeList(f, _)
780 | DataType::LargeList(f)
781 | DataType::ListView(f)
782 | DataType::LargeListView(f) => {
783 let field_ref = if coerce_types && f.name() != PARQUET_LIST_ELEMENT_NAME {
784 let ff = f.as_ref().clone().with_name(PARQUET_LIST_ELEMENT_NAME);
786 Arc::new(arrow_to_parquet_type(&ff, coerce_types)?)
787 } else {
788 Arc::new(arrow_to_parquet_type(f, coerce_types)?)
789 };
790
791 Type::group_type_builder(name)
792 .with_fields(vec![Arc::new(
793 Type::group_type_builder("list")
794 .with_fields(vec![field_ref])
795 .with_repetition(Repetition::REPEATED)
796 .build()?,
797 )])
798 .with_logical_type(Some(LogicalType::List))
799 .with_repetition(repetition)
800 .with_id(id)
801 .build()
802 }
803 DataType::Struct(fields) => {
804 if fields.is_empty() {
805 return Err(arrow_err!("Parquet does not support writing empty structs",));
806 }
807 let fields = fields
809 .iter()
810 .map(|f| arrow_to_parquet_type(f, coerce_types).map(Arc::new))
811 .collect::<Result<_>>()?;
812 Type::group_type_builder(name)
813 .with_fields(fields)
814 .with_repetition(repetition)
815 .with_id(id)
816 .with_logical_type(logical_type_for_struct(field))
817 .build()
818 }
819 DataType::Map(field, _) => {
820 if let DataType::Struct(struct_fields) = field.data_type() {
821 let map_struct_name = if coerce_types {
823 PARQUET_MAP_STRUCT_NAME
824 } else {
825 field.name()
826 };
827
828 let fix_map_field = |name: &str, fld: &Arc<Field>| -> Result<Arc<Type>> {
830 if coerce_types && fld.name() != name {
831 let f = fld.as_ref().clone().with_name(name);
832 Ok(Arc::new(arrow_to_parquet_type(&f, coerce_types)?))
833 } else {
834 Ok(Arc::new(arrow_to_parquet_type(fld, coerce_types)?))
835 }
836 };
837 let key_field = fix_map_field(PARQUET_KEY_FIELD_NAME, &struct_fields[0])?;
838 let val_field = fix_map_field(PARQUET_VALUE_FIELD_NAME, &struct_fields[1])?;
839
840 Type::group_type_builder(name)
841 .with_fields(vec![Arc::new(
842 Type::group_type_builder(map_struct_name)
843 .with_fields(vec![key_field, val_field])
844 .with_repetition(Repetition::REPEATED)
845 .build()?,
846 )])
847 .with_logical_type(Some(LogicalType::Map))
848 .with_repetition(repetition)
849 .with_id(id)
850 .build()
851 } else {
852 Err(arrow_err!(
853 "DataType::Map should contain a struct field child",
854 ))
855 }
856 }
857 DataType::Union(_, _) => unimplemented!("See ARROW-8817."),
858 DataType::Dictionary(_, value) => {
859 let dict_field = field.clone().with_data_type(value.as_ref().clone());
861 arrow_to_parquet_type(&dict_field, coerce_types)
862 }
863 DataType::RunEndEncoded(_, value_field) => {
864 let ree_value_field = field
865 .clone()
866 .with_data_type(value_field.data_type().clone());
867 arrow_to_parquet_type(&ree_value_field, coerce_types)
868 }
869 }
870}
871
872fn field_id(field: &Field) -> Option<i32> {
873 let value = field.metadata().get(super::PARQUET_FIELD_ID_META_KEY)?;
874 value.parse().ok() }
876
877#[cfg(test)]
878mod tests {
879 use super::*;
880
881 use std::{collections::HashMap, sync::Arc};
882
883 use crate::arrow::PARQUET_FIELD_ID_META_KEY;
884 use crate::file::metadata::KeyValue;
885 use crate::file::reader::FileReader;
886 use crate::{
887 arrow::{ArrowWriter, arrow_reader::ParquetRecordBatchReaderBuilder},
888 schema::{parser::parse_message_type, types::SchemaDescriptor},
889 };
890 use arrow::datatypes::{DataType, Field, IntervalUnit, TimeUnit};
891
892 #[test]
893 fn test_flat_primitives() {
894 let message_type = "
895 message test_schema {
896 REQUIRED BOOLEAN boolean;
897 REQUIRED INT32 int8 (INT_8);
898 REQUIRED INT32 int16 (INT_16);
899 REQUIRED INT32 uint8 (INTEGER(8,false));
900 REQUIRED INT32 uint16 (INTEGER(16,false));
901 REQUIRED INT32 int32;
902 REQUIRED INT64 int64;
903 OPTIONAL DOUBLE double;
904 OPTIONAL FLOAT float;
905 OPTIONAL FIXED_LEN_BYTE_ARRAY (2) float16 (FLOAT16);
906 OPTIONAL BINARY string (UTF8);
907 OPTIONAL BINARY string_2 (STRING);
908 OPTIONAL BINARY json (JSON);
909 }
910 ";
911 let parquet_group_type = parse_message_type(message_type).unwrap();
912
913 let parquet_schema = SchemaDescriptor::new(Arc::new(parquet_group_type));
914 let converted_arrow_schema = parquet_to_arrow_schema(&parquet_schema, None).unwrap();
915
916 let arrow_fields = Fields::from(vec![
917 Field::new("boolean", DataType::Boolean, false),
918 Field::new("int8", DataType::Int8, false),
919 Field::new("int16", DataType::Int16, false),
920 Field::new("uint8", DataType::UInt8, false),
921 Field::new("uint16", DataType::UInt16, false),
922 Field::new("int32", DataType::Int32, false),
923 Field::new("int64", DataType::Int64, false),
924 Field::new("double", DataType::Float64, true),
925 Field::new("float", DataType::Float32, true),
926 Field::new("float16", DataType::Float16, true),
927 Field::new("string", DataType::Utf8, true),
928 Field::new("string_2", DataType::Utf8, true),
929 json_field(),
930 ]);
931
932 assert_eq!(&arrow_fields, converted_arrow_schema.fields());
933 }
934
935 fn json_field() -> Field {
938 #[cfg(feature = "arrow_canonical_extension_types")]
939 {
940 Field::new("json", DataType::Utf8, true)
941 .with_extension_type(arrow_schema::extension::Json::default())
942 }
943 #[cfg(not(feature = "arrow_canonical_extension_types"))]
944 {
945 Field::new("json", DataType::Utf8, true)
946 }
947 }
948
949 #[test]
950 fn test_decimal_fields() {
951 let message_type = "
952 message test_schema {
953 REQUIRED INT32 decimal1 (DECIMAL(4,2));
954 REQUIRED INT64 decimal2 (DECIMAL(12,2));
955 REQUIRED FIXED_LEN_BYTE_ARRAY (16) decimal3 (DECIMAL(30,2));
956 REQUIRED BYTE_ARRAY decimal4 (DECIMAL(33,2));
957 REQUIRED BYTE_ARRAY decimal5 (DECIMAL(38,2));
958 REQUIRED FIXED_LEN_BYTE_ARRAY (17) decimal6 (DECIMAL(39,2));
959 REQUIRED BYTE_ARRAY decimal7 (DECIMAL(39,2));
960 }
961 ";
962
963 let parquet_group_type = parse_message_type(message_type).unwrap();
964
965 let parquet_schema = SchemaDescriptor::new(Arc::new(parquet_group_type));
966 let converted_arrow_schema = parquet_to_arrow_schema(&parquet_schema, None).unwrap();
967
968 let arrow_fields = Fields::from(vec![
969 Field::new("decimal1", DataType::Decimal128(4, 2), false),
970 Field::new("decimal2", DataType::Decimal128(12, 2), false),
971 Field::new("decimal3", DataType::Decimal128(30, 2), false),
972 Field::new("decimal4", DataType::Decimal128(33, 2), false),
973 Field::new("decimal5", DataType::Decimal128(38, 2), false),
974 Field::new("decimal6", DataType::Decimal256(39, 2), false),
975 Field::new("decimal7", DataType::Decimal256(39, 2), false),
976 ]);
977 assert_eq!(&arrow_fields, converted_arrow_schema.fields());
978 }
979
980 #[test]
981 fn test_byte_array_fields() {
982 let message_type = "
983 message test_schema {
984 REQUIRED BYTE_ARRAY binary;
985 REQUIRED FIXED_LEN_BYTE_ARRAY (20) fixed_binary;
986 }
987 ";
988
989 let parquet_group_type = parse_message_type(message_type).unwrap();
990
991 let parquet_schema = SchemaDescriptor::new(Arc::new(parquet_group_type));
992 let converted_arrow_schema = parquet_to_arrow_schema(&parquet_schema, None).unwrap();
993
994 let arrow_fields = Fields::from(vec![
995 Field::new("binary", DataType::Binary, false),
996 Field::new("fixed_binary", DataType::FixedSizeBinary(20), false),
997 ]);
998 assert_eq!(&arrow_fields, converted_arrow_schema.fields());
999 }
1000
1001 #[test]
1002 fn test_duplicate_fields() {
1003 let message_type = "
1004 message test_schema {
1005 REQUIRED BOOLEAN boolean;
1006 REQUIRED INT32 int8 (INT_8);
1007 }
1008 ";
1009
1010 let parquet_group_type = parse_message_type(message_type).unwrap();
1011
1012 let parquet_schema = SchemaDescriptor::new(Arc::new(parquet_group_type));
1013 let converted_arrow_schema = parquet_to_arrow_schema(&parquet_schema, None).unwrap();
1014
1015 let arrow_fields = Fields::from(vec![
1016 Field::new("boolean", DataType::Boolean, false),
1017 Field::new("int8", DataType::Int8, false),
1018 ]);
1019 assert_eq!(&arrow_fields, converted_arrow_schema.fields());
1020
1021 let converted_arrow_schema =
1022 parquet_to_arrow_schema_by_columns(&parquet_schema, ProjectionMask::all(), None)
1023 .unwrap();
1024 assert_eq!(&arrow_fields, converted_arrow_schema.fields());
1025 }
1026
1027 #[test]
1028 fn test_parquet_lists() {
1029 let mut arrow_fields = Vec::new();
1030
1031 let message_type = "
1033 message test_schema {
1034 REQUIRED GROUP my_list (LIST) {
1035 REPEATED GROUP list {
1036 OPTIONAL BINARY element (UTF8);
1037 }
1038 }
1039 OPTIONAL GROUP my_list (LIST) {
1040 REPEATED GROUP list {
1041 REQUIRED BINARY element (UTF8);
1042 }
1043 }
1044 OPTIONAL GROUP array_of_arrays (LIST) {
1045 REPEATED GROUP list {
1046 REQUIRED GROUP element (LIST) {
1047 REPEATED GROUP list {
1048 REQUIRED INT32 element;
1049 }
1050 }
1051 }
1052 }
1053 OPTIONAL GROUP my_list (LIST) {
1054 REPEATED GROUP element {
1055 REQUIRED BINARY str (UTF8);
1056 }
1057 }
1058 OPTIONAL GROUP my_list (LIST) {
1059 REPEATED INT32 element;
1060 }
1061 OPTIONAL GROUP my_list (LIST) {
1062 REPEATED GROUP element {
1063 REQUIRED BINARY str (UTF8);
1064 REQUIRED INT32 num;
1065 }
1066 }
1067 OPTIONAL GROUP my_list (LIST) {
1068 REPEATED GROUP array {
1069 REQUIRED BINARY str (UTF8);
1070 }
1071
1072 }
1073 OPTIONAL GROUP my_list (LIST) {
1074 REPEATED GROUP my_list_tuple {
1075 REQUIRED BINARY str (UTF8);
1076 }
1077 }
1078 REPEATED INT32 name;
1079 }
1080 ";
1081
1082 {
1089 arrow_fields.push(Field::new_list(
1090 "my_list",
1091 Field::new("element", DataType::Utf8, true),
1092 false,
1093 ));
1094 }
1095
1096 {
1103 arrow_fields.push(Field::new_list(
1104 "my_list",
1105 Field::new("element", DataType::Utf8, false),
1106 true,
1107 ));
1108 }
1109
1110 {
1123 let arrow_inner_list = Field::new("element", DataType::Int32, false);
1124 arrow_fields.push(Field::new_list(
1125 "array_of_arrays",
1126 Field::new_list("element", arrow_inner_list, false),
1127 true,
1128 ));
1129 }
1130
1131 {
1138 arrow_fields.push(Field::new_list(
1139 "my_list",
1140 Field::new("str", DataType::Utf8, false),
1141 true,
1142 ));
1143 }
1144
1145 {
1150 arrow_fields.push(Field::new_list(
1151 "my_list",
1152 Field::new("element", DataType::Int32, false),
1153 true,
1154 ));
1155 }
1156
1157 {
1165 let fields = vec![
1166 Field::new("str", DataType::Utf8, false),
1167 Field::new("num", DataType::Int32, false),
1168 ];
1169 arrow_fields.push(Field::new_list(
1170 "my_list",
1171 Field::new_struct("element", fields, false),
1172 true,
1173 ));
1174 }
1175
1176 {
1184 let fields = vec![Field::new("str", DataType::Utf8, false)];
1185 arrow_fields.push(Field::new_list(
1186 "my_list",
1187 Field::new_struct("array", fields, false),
1188 true,
1189 ));
1190 }
1191
1192 {
1200 let fields = vec![Field::new("str", DataType::Utf8, false)];
1201 arrow_fields.push(Field::new_list(
1202 "my_list",
1203 Field::new_struct("my_list_tuple", fields, false),
1204 true,
1205 ));
1206 }
1207
1208 {
1211 arrow_fields.push(Field::new_list(
1212 "name",
1213 Field::new("name", DataType::Int32, false),
1214 false,
1215 ));
1216 }
1217
1218 let parquet_group_type = parse_message_type(message_type).unwrap();
1219
1220 let parquet_schema = SchemaDescriptor::new(Arc::new(parquet_group_type));
1221 let converted_arrow_schema = parquet_to_arrow_schema(&parquet_schema, None).unwrap();
1222 let converted_fields = converted_arrow_schema.fields();
1223
1224 assert_eq!(arrow_fields.len(), converted_fields.len());
1225 for i in 0..arrow_fields.len() {
1226 assert_eq!(&arrow_fields[i], converted_fields[i].as_ref(), "{i}");
1227 }
1228 }
1229
1230 #[test]
1231 fn test_parquet_list_nullable() {
1232 let mut arrow_fields = Vec::new();
1233
1234 let message_type = "
1235 message test_schema {
1236 REQUIRED GROUP my_list1 (LIST) {
1237 REPEATED GROUP list {
1238 OPTIONAL BINARY element (UTF8);
1239 }
1240 }
1241 OPTIONAL GROUP my_list2 (LIST) {
1242 REPEATED GROUP list {
1243 REQUIRED BINARY element (UTF8);
1244 }
1245 }
1246 REQUIRED GROUP my_list3 (LIST) {
1247 REPEATED GROUP list {
1248 REQUIRED BINARY element (UTF8);
1249 }
1250 }
1251 }
1252 ";
1253
1254 {
1261 arrow_fields.push(Field::new_list(
1262 "my_list1",
1263 Field::new("element", DataType::Utf8, true),
1264 false,
1265 ));
1266 }
1267
1268 {
1275 arrow_fields.push(Field::new_list(
1276 "my_list2",
1277 Field::new("element", DataType::Utf8, false),
1278 true,
1279 ));
1280 }
1281
1282 {
1289 arrow_fields.push(Field::new_list(
1290 "my_list3",
1291 Field::new("element", DataType::Utf8, false),
1292 false,
1293 ));
1294 }
1295
1296 let parquet_group_type = parse_message_type(message_type).unwrap();
1297
1298 let parquet_schema = SchemaDescriptor::new(Arc::new(parquet_group_type));
1299 let converted_arrow_schema = parquet_to_arrow_schema(&parquet_schema, None).unwrap();
1300 let converted_fields = converted_arrow_schema.fields();
1301
1302 assert_eq!(arrow_fields.len(), converted_fields.len());
1303 for i in 0..arrow_fields.len() {
1304 assert_eq!(&arrow_fields[i], converted_fields[i].as_ref());
1305 }
1306 }
1307
1308 #[test]
1309 fn test_parquet_maps() {
1310 let mut arrow_fields = Vec::new();
1311
1312 let message_type = "
1314 message test_schema {
1315 REQUIRED group my_map1 (MAP) {
1316 REPEATED group key_value {
1317 REQUIRED binary key (UTF8);
1318 OPTIONAL int32 value;
1319 }
1320 }
1321 OPTIONAL group my_map2 (MAP) {
1322 REPEATED group map {
1323 REQUIRED binary str (UTF8);
1324 REQUIRED int32 num;
1325 }
1326 }
1327 OPTIONAL group my_map3 (MAP_KEY_VALUE) {
1328 REPEATED group map {
1329 REQUIRED binary key (UTF8);
1330 OPTIONAL int32 value;
1331 }
1332 }
1333 REQUIRED group my_map4 (MAP) {
1334 REPEATED group map {
1335 OPTIONAL binary key (UTF8);
1336 REQUIRED int32 value;
1337 }
1338 }
1339 }
1340 ";
1341
1342 {
1350 arrow_fields.push(Field::new_map(
1351 "my_map1",
1352 "key_value",
1353 Field::new("key", DataType::Utf8, false),
1354 Field::new("value", DataType::Int32, true),
1355 false,
1356 false,
1357 ));
1358 }
1359
1360 {
1368 arrow_fields.push(Field::new_map(
1369 "my_map2",
1370 "map",
1371 Field::new("str", DataType::Utf8, false),
1372 Field::new("num", DataType::Int32, false),
1373 false,
1374 true,
1375 ));
1376 }
1377
1378 {
1386 arrow_fields.push(Field::new_map(
1387 "my_map3",
1388 "map",
1389 Field::new("key", DataType::Utf8, false),
1390 Field::new("value", DataType::Int32, true),
1391 false,
1392 true,
1393 ));
1394 }
1395
1396 {
1404 arrow_fields.push(Field::new_map(
1405 "my_map4",
1406 "map",
1407 Field::new("key", DataType::Utf8, false), Field::new("value", DataType::Int32, false),
1409 false,
1410 false,
1411 ));
1412 }
1413
1414 let parquet_group_type = parse_message_type(message_type).unwrap();
1415
1416 let parquet_schema = SchemaDescriptor::new(Arc::new(parquet_group_type));
1417 let converted_arrow_schema = parquet_to_arrow_schema(&parquet_schema, None).unwrap();
1418 let converted_fields = converted_arrow_schema.fields();
1419
1420 assert_eq!(arrow_fields.len(), converted_fields.len());
1421 for i in 0..arrow_fields.len() {
1422 assert_eq!(&arrow_fields[i], converted_fields[i].as_ref());
1423 }
1424 }
1425
1426 #[test]
1427 fn test_nested_schema() {
1428 let mut arrow_fields = Vec::new();
1429 {
1430 let group1_fields = Fields::from(vec![
1431 Field::new("leaf1", DataType::Boolean, false),
1432 Field::new("leaf2", DataType::Int32, false),
1433 ]);
1434 let group1_struct = Field::new("group1", DataType::Struct(group1_fields), false);
1435 arrow_fields.push(group1_struct);
1436
1437 let leaf3_field = Field::new("leaf3", DataType::Int64, false);
1438 arrow_fields.push(leaf3_field);
1439 }
1440
1441 let message_type = "
1442 message test_schema {
1443 REQUIRED GROUP group1 {
1444 REQUIRED BOOLEAN leaf1;
1445 REQUIRED INT32 leaf2;
1446 }
1447 REQUIRED INT64 leaf3;
1448 }
1449 ";
1450 let parquet_group_type = parse_message_type(message_type).unwrap();
1451
1452 let parquet_schema = SchemaDescriptor::new(Arc::new(parquet_group_type));
1453 let converted_arrow_schema = parquet_to_arrow_schema(&parquet_schema, None).unwrap();
1454 let converted_fields = converted_arrow_schema.fields();
1455
1456 assert_eq!(arrow_fields.len(), converted_fields.len());
1457 for i in 0..arrow_fields.len() {
1458 assert_eq!(&arrow_fields[i], converted_fields[i].as_ref());
1459 }
1460 }
1461
1462 #[test]
1463 fn test_nested_schema_partial() {
1464 let mut arrow_fields = Vec::new();
1465 {
1466 let group1_fields = vec![Field::new("leaf1", DataType::Int64, false)].into();
1467 let group1 = Field::new("group1", DataType::Struct(group1_fields), false);
1468 arrow_fields.push(group1);
1469
1470 let group2_fields = vec![Field::new("leaf4", DataType::Int64, false)].into();
1471 let group2 = Field::new("group2", DataType::Struct(group2_fields), false);
1472 arrow_fields.push(group2);
1473
1474 arrow_fields.push(Field::new("leaf5", DataType::Int64, false));
1475 }
1476
1477 let message_type = "
1478 message test_schema {
1479 REQUIRED GROUP group1 {
1480 REQUIRED INT64 leaf1;
1481 REQUIRED INT64 leaf2;
1482 }
1483 REQUIRED GROUP group2 {
1484 REQUIRED INT64 leaf3;
1485 REQUIRED INT64 leaf4;
1486 }
1487 REQUIRED INT64 leaf5;
1488 }
1489 ";
1490 let parquet_group_type = parse_message_type(message_type).unwrap();
1491
1492 let parquet_schema = SchemaDescriptor::new(Arc::new(parquet_group_type));
1502 let mask = ProjectionMask::leaves(&parquet_schema, [3, 0, 4, 4]);
1503 let converted_arrow_schema =
1504 parquet_to_arrow_schema_by_columns(&parquet_schema, mask, None).unwrap();
1505 let converted_fields = converted_arrow_schema.fields();
1506
1507 assert_eq!(arrow_fields.len(), converted_fields.len());
1508 for i in 0..arrow_fields.len() {
1509 assert_eq!(&arrow_fields[i], converted_fields[i].as_ref());
1510 }
1511 }
1512
1513 #[test]
1514 fn test_nested_schema_partial_ordering() {
1515 let mut arrow_fields = Vec::new();
1516 {
1517 let group1_fields = vec![Field::new("leaf1", DataType::Int64, false)].into();
1518 let group1 = Field::new("group1", DataType::Struct(group1_fields), false);
1519 arrow_fields.push(group1);
1520
1521 let group2_fields = vec![Field::new("leaf4", DataType::Int64, false)].into();
1522 let group2 = Field::new("group2", DataType::Struct(group2_fields), false);
1523 arrow_fields.push(group2);
1524
1525 arrow_fields.push(Field::new("leaf5", DataType::Int64, false));
1526 }
1527
1528 let message_type = "
1529 message test_schema {
1530 REQUIRED GROUP group1 {
1531 REQUIRED INT64 leaf1;
1532 REQUIRED INT64 leaf2;
1533 }
1534 REQUIRED GROUP group2 {
1535 REQUIRED INT64 leaf3;
1536 REQUIRED INT64 leaf4;
1537 }
1538 REQUIRED INT64 leaf5;
1539 }
1540 ";
1541 let parquet_group_type = parse_message_type(message_type).unwrap();
1542
1543 let parquet_schema = SchemaDescriptor::new(Arc::new(parquet_group_type));
1553 let mask = ProjectionMask::leaves(&parquet_schema, [3, 0, 4]);
1554 let converted_arrow_schema =
1555 parquet_to_arrow_schema_by_columns(&parquet_schema, mask, None).unwrap();
1556 let converted_fields = converted_arrow_schema.fields();
1557
1558 assert_eq!(arrow_fields.len(), converted_fields.len());
1559 for i in 0..arrow_fields.len() {
1560 assert_eq!(&arrow_fields[i], converted_fields[i].as_ref());
1561 }
1562
1563 let mask =
1564 ProjectionMask::columns(&parquet_schema, ["group2.leaf4", "group1.leaf1", "leaf5"]);
1565 let converted_arrow_schema =
1566 parquet_to_arrow_schema_by_columns(&parquet_schema, mask, None).unwrap();
1567 let converted_fields = converted_arrow_schema.fields();
1568
1569 assert_eq!(arrow_fields.len(), converted_fields.len());
1570 for i in 0..arrow_fields.len() {
1571 assert_eq!(&arrow_fields[i], converted_fields[i].as_ref());
1572 }
1573 }
1574
1575 #[test]
1576 fn test_repeated_nested_schema() {
1577 let mut arrow_fields = Vec::new();
1578 {
1579 arrow_fields.push(Field::new("leaf1", DataType::Int32, true));
1580
1581 let inner_group_list = Field::new_list(
1582 "innerGroup",
1583 Field::new_struct(
1584 "innerGroup",
1585 vec![Field::new("leaf3", DataType::Int32, true)],
1586 false,
1587 ),
1588 false,
1589 );
1590
1591 let outer_group_list = Field::new_list(
1592 "outerGroup",
1593 Field::new_struct(
1594 "outerGroup",
1595 vec![Field::new("leaf2", DataType::Int32, true), inner_group_list],
1596 false,
1597 ),
1598 false,
1599 );
1600 arrow_fields.push(outer_group_list);
1601 }
1602
1603 let message_type = "
1604 message test_schema {
1605 OPTIONAL INT32 leaf1;
1606 REPEATED GROUP outerGroup {
1607 OPTIONAL INT32 leaf2;
1608 REPEATED GROUP innerGroup {
1609 OPTIONAL INT32 leaf3;
1610 }
1611 }
1612 }
1613 ";
1614 let parquet_group_type = parse_message_type(message_type).unwrap();
1615
1616 let parquet_schema = SchemaDescriptor::new(Arc::new(parquet_group_type));
1617 let converted_arrow_schema = parquet_to_arrow_schema(&parquet_schema, None).unwrap();
1618 let converted_fields = converted_arrow_schema.fields();
1619
1620 assert_eq!(arrow_fields.len(), converted_fields.len());
1621 for i in 0..arrow_fields.len() {
1622 assert_eq!(&arrow_fields[i], converted_fields[i].as_ref());
1623 }
1624 }
1625
1626 #[test]
1627 fn test_column_desc_to_field() {
1628 let message_type = "
1629 message test_schema {
1630 REQUIRED BOOLEAN boolean;
1631 REQUIRED INT32 int8 (INT_8);
1632 REQUIRED INT32 uint8 (INTEGER(8,false));
1633 REQUIRED INT32 int16 (INT_16);
1634 REQUIRED INT32 uint16 (INTEGER(16,false));
1635 REQUIRED INT32 int32;
1636 REQUIRED INT64 int64;
1637 OPTIONAL DOUBLE double;
1638 OPTIONAL FLOAT float;
1639 OPTIONAL FIXED_LEN_BYTE_ARRAY (2) float16 (FLOAT16);
1640 OPTIONAL BINARY string (UTF8);
1641 REPEATED BOOLEAN bools;
1642 OPTIONAL INT32 date (DATE);
1643 OPTIONAL INT32 time_milli (TIME_MILLIS);
1644 OPTIONAL INT64 time_micro (TIME_MICROS);
1645 OPTIONAL INT64 time_nano (TIME(NANOS,false));
1646 OPTIONAL INT64 ts_milli (TIMESTAMP_MILLIS);
1647 REQUIRED INT64 ts_micro (TIMESTAMP_MICROS);
1648 REQUIRED INT64 ts_nano (TIMESTAMP(NANOS,true));
1649 REPEATED INT32 int_list;
1650 REPEATED BINARY byte_list;
1651 REPEATED BINARY string_list (UTF8);
1652 REQUIRED INT32 decimal_int32 (DECIMAL(8,2));
1653 REQUIRED INT64 decimal_int64 (DECIMAL(16,2));
1654 REQUIRED FIXED_LEN_BYTE_ARRAY (13) decimal_fix_length (DECIMAL(30,2));
1655 }
1656 ";
1657 let parquet_group_type = parse_message_type(message_type).unwrap();
1658
1659 let parquet_schema = SchemaDescriptor::new(Arc::new(parquet_group_type));
1660 let converted_arrow_fields = parquet_schema
1661 .columns()
1662 .iter()
1663 .map(|c| parquet_to_arrow_field(c).unwrap())
1664 .collect::<Vec<Field>>();
1665
1666 let arrow_fields = vec![
1667 Field::new("boolean", DataType::Boolean, false),
1668 Field::new("int8", DataType::Int8, false),
1669 Field::new("uint8", DataType::UInt8, false),
1670 Field::new("int16", DataType::Int16, false),
1671 Field::new("uint16", DataType::UInt16, false),
1672 Field::new("int32", DataType::Int32, false),
1673 Field::new("int64", DataType::Int64, false),
1674 Field::new("double", DataType::Float64, true),
1675 Field::new("float", DataType::Float32, true),
1676 Field::new("float16", DataType::Float16, true),
1677 Field::new("string", DataType::Utf8, true),
1678 Field::new_list(
1679 "bools",
1680 Field::new("bools", DataType::Boolean, false),
1681 false,
1682 ),
1683 Field::new("date", DataType::Date32, true),
1684 Field::new("time_milli", DataType::Time32(TimeUnit::Millisecond), true),
1685 Field::new("time_micro", DataType::Time64(TimeUnit::Microsecond), true),
1686 Field::new("time_nano", DataType::Time64(TimeUnit::Nanosecond), true),
1687 Field::new(
1688 "ts_milli",
1689 DataType::Timestamp(TimeUnit::Millisecond, Some("UTC".into())),
1690 true,
1691 ),
1692 Field::new(
1693 "ts_micro",
1694 DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into())),
1695 false,
1696 ),
1697 Field::new(
1698 "ts_nano",
1699 DataType::Timestamp(TimeUnit::Nanosecond, Some("UTC".into())),
1700 false,
1701 ),
1702 Field::new_list(
1703 "int_list",
1704 Field::new("int_list", DataType::Int32, false),
1705 false,
1706 ),
1707 Field::new_list(
1708 "byte_list",
1709 Field::new("byte_list", DataType::Binary, false),
1710 false,
1711 ),
1712 Field::new_list(
1713 "string_list",
1714 Field::new("string_list", DataType::Utf8, false),
1715 false,
1716 ),
1717 Field::new("decimal_int32", DataType::Decimal128(8, 2), false),
1718 Field::new("decimal_int64", DataType::Decimal128(16, 2), false),
1719 Field::new("decimal_fix_length", DataType::Decimal128(30, 2), false),
1720 ];
1721
1722 assert_eq!(arrow_fields, converted_arrow_fields);
1723 }
1724
1725 #[test]
1726 fn test_coerced_map_list() {
1727 let arrow_fields = vec![
1729 Field::new_list(
1730 "my_list",
1731 Field::new("item", DataType::Boolean, true),
1732 false,
1733 ),
1734 Field::new_map(
1735 "my_map",
1736 "my_entries",
1737 Field::new("my_keys", DataType::Utf8, false),
1738 Field::new("my_values", DataType::Int32, true),
1739 false,
1740 true,
1741 ),
1742 ];
1743 let arrow_schema = Schema::new(arrow_fields);
1744
1745 let message_type = "
1747 message parquet_schema {
1748 REQUIRED GROUP my_list (LIST) {
1749 REPEATED GROUP list {
1750 OPTIONAL BOOLEAN element;
1751 }
1752 }
1753 OPTIONAL GROUP my_map (MAP) {
1754 REPEATED GROUP key_value {
1755 REQUIRED BINARY key (STRING);
1756 OPTIONAL INT32 value;
1757 }
1758 }
1759 }
1760 ";
1761 let parquet_group_type = parse_message_type(message_type).unwrap();
1762 let parquet_schema = SchemaDescriptor::new(Arc::new(parquet_group_type));
1763 let converted_arrow_schema = ArrowSchemaConverter::new()
1764 .with_coerce_types(true)
1765 .convert(&arrow_schema)
1766 .unwrap();
1767 assert_eq!(
1768 parquet_schema.columns().len(),
1769 converted_arrow_schema.columns().len()
1770 );
1771
1772 let message_type = "
1774 message parquet_schema {
1775 REQUIRED GROUP my_list (LIST) {
1776 REPEATED GROUP list {
1777 OPTIONAL BOOLEAN item;
1778 }
1779 }
1780 OPTIONAL GROUP my_map (MAP) {
1781 REPEATED GROUP my_entries {
1782 REQUIRED BINARY my_keys (STRING);
1783 OPTIONAL INT32 my_values;
1784 }
1785 }
1786 }
1787 ";
1788 let parquet_group_type = parse_message_type(message_type).unwrap();
1789 let parquet_schema = SchemaDescriptor::new(Arc::new(parquet_group_type));
1790 let converted_arrow_schema = ArrowSchemaConverter::new()
1791 .with_coerce_types(false)
1792 .convert(&arrow_schema)
1793 .unwrap();
1794 assert_eq!(
1795 parquet_schema.columns().len(),
1796 converted_arrow_schema.columns().len()
1797 );
1798 }
1799
1800 #[test]
1801 fn test_field_to_column_desc() {
1802 let message_type = "
1803 message arrow_schema {
1804 REQUIRED BOOLEAN boolean;
1805 REQUIRED INT32 int8 (INT_8);
1806 REQUIRED INT32 int16 (INTEGER(16,true));
1807 REQUIRED INT32 int32;
1808 REQUIRED INT64 int64;
1809 OPTIONAL DOUBLE double;
1810 OPTIONAL FLOAT float;
1811 OPTIONAL FIXED_LEN_BYTE_ARRAY (2) float16 (FLOAT16);
1812 OPTIONAL BINARY string (STRING);
1813 OPTIONAL GROUP bools (LIST) {
1814 REPEATED GROUP list {
1815 OPTIONAL BOOLEAN element;
1816 }
1817 }
1818 REQUIRED GROUP bools_non_null (LIST) {
1819 REPEATED GROUP list {
1820 REQUIRED BOOLEAN element;
1821 }
1822 }
1823 OPTIONAL INT32 date (DATE);
1824 OPTIONAL INT32 time_milli (TIME(MILLIS,false));
1825 OPTIONAL INT32 time_milli_utc (TIME(MILLIS,true));
1826 OPTIONAL INT64 time_micro (TIME_MICROS);
1827 OPTIONAL INT64 time_micro_utc (TIME(MICROS, true));
1828 OPTIONAL INT64 ts_milli (TIMESTAMP_MILLIS);
1829 REQUIRED INT64 ts_micro (TIMESTAMP(MICROS,false));
1830 REQUIRED INT64 ts_seconds;
1831 REQUIRED INT64 ts_micro_utc (TIMESTAMP(MICROS, true));
1832 REQUIRED INT64 ts_millis_zero_offset (TIMESTAMP(MILLIS, true));
1833 REQUIRED INT64 ts_millis_zero_negative_offset (TIMESTAMP(MILLIS, true));
1834 REQUIRED INT64 ts_micro_non_utc (TIMESTAMP(MICROS, true));
1835 REQUIRED GROUP struct {
1836 REQUIRED BOOLEAN bools;
1837 REQUIRED INT32 uint32 (INTEGER(32,false));
1838 REQUIRED GROUP int32 (LIST) {
1839 REPEATED GROUP list {
1840 OPTIONAL INT32 element;
1841 }
1842 }
1843 }
1844 REQUIRED BINARY dictionary_strings (STRING);
1845 REQUIRED INT32 decimal_int32 (DECIMAL(8,2));
1846 REQUIRED INT64 decimal_int64 (DECIMAL(16,2));
1847 REQUIRED FIXED_LEN_BYTE_ARRAY (13) decimal_fix_length (DECIMAL(30,2));
1848 REQUIRED FIXED_LEN_BYTE_ARRAY (16) decimal128 (DECIMAL(38,2));
1849 REQUIRED FIXED_LEN_BYTE_ARRAY (17) decimal256 (DECIMAL(39,2));
1850 }
1851 ";
1852 let parquet_group_type = parse_message_type(message_type).unwrap();
1853
1854 let parquet_schema = SchemaDescriptor::new(Arc::new(parquet_group_type));
1855
1856 let arrow_fields = vec![
1857 Field::new("boolean", DataType::Boolean, false),
1858 Field::new("int8", DataType::Int8, false),
1859 Field::new("int16", DataType::Int16, false),
1860 Field::new("int32", DataType::Int32, false),
1861 Field::new("int64", DataType::Int64, false),
1862 Field::new("double", DataType::Float64, true),
1863 Field::new("float", DataType::Float32, true),
1864 Field::new("float16", DataType::Float16, true),
1865 Field::new("string", DataType::Utf8, true),
1866 Field::new_list(
1867 "bools",
1868 Field::new("element", DataType::Boolean, true),
1869 true,
1870 ),
1871 Field::new_list(
1872 "bools_non_null",
1873 Field::new("element", DataType::Boolean, false),
1874 false,
1875 ),
1876 Field::new("date", DataType::Date32, true),
1877 Field::new("time_milli", DataType::Time32(TimeUnit::Millisecond), true),
1878 Field::new(
1879 "time_milli_utc",
1880 DataType::Time32(TimeUnit::Millisecond),
1881 true,
1882 )
1883 .with_metadata(HashMap::from_iter(vec![(
1884 "adjusted_to_utc".to_string(),
1885 String::new(),
1886 )])),
1887 Field::new("time_micro", DataType::Time64(TimeUnit::Microsecond), true),
1888 Field::new(
1889 "time_micro_utc",
1890 DataType::Time64(TimeUnit::Microsecond),
1891 true,
1892 )
1893 .with_metadata(HashMap::from_iter(vec![(
1894 "adjusted_to_utc".to_string(),
1895 String::new(),
1896 )])),
1897 Field::new(
1898 "ts_milli",
1899 DataType::Timestamp(TimeUnit::Millisecond, None),
1900 true,
1901 ),
1902 Field::new(
1903 "ts_micro",
1904 DataType::Timestamp(TimeUnit::Microsecond, None),
1905 false,
1906 ),
1907 Field::new(
1908 "ts_seconds",
1909 DataType::Timestamp(TimeUnit::Second, Some("UTC".into())),
1910 false,
1911 ),
1912 Field::new(
1913 "ts_micro_utc",
1914 DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into())),
1915 false,
1916 ),
1917 Field::new(
1918 "ts_millis_zero_offset",
1919 DataType::Timestamp(TimeUnit::Millisecond, Some("+00:00".into())),
1920 false,
1921 ),
1922 Field::new(
1923 "ts_millis_zero_negative_offset",
1924 DataType::Timestamp(TimeUnit::Millisecond, Some("-00:00".into())),
1925 false,
1926 ),
1927 Field::new(
1928 "ts_micro_non_utc",
1929 DataType::Timestamp(TimeUnit::Microsecond, Some("+01:00".into())),
1930 false,
1931 ),
1932 Field::new_struct(
1933 "struct",
1934 vec![
1935 Field::new("bools", DataType::Boolean, false),
1936 Field::new("uint32", DataType::UInt32, false),
1937 Field::new_list("int32", Field::new("element", DataType::Int32, true), false),
1938 ],
1939 false,
1940 ),
1941 Field::new_dictionary("dictionary_strings", DataType::Int32, DataType::Utf8, false),
1942 Field::new("decimal_int32", DataType::Decimal128(8, 2), false),
1943 Field::new("decimal_int64", DataType::Decimal128(16, 2), false),
1944 Field::new("decimal_fix_length", DataType::Decimal128(30, 2), false),
1945 Field::new("decimal128", DataType::Decimal128(38, 2), false),
1946 Field::new("decimal256", DataType::Decimal256(39, 2), false),
1947 ];
1948 let arrow_schema = Schema::new(arrow_fields);
1949 let converted_arrow_schema = ArrowSchemaConverter::new().convert(&arrow_schema).unwrap();
1950
1951 assert_eq!(
1952 parquet_schema.columns().len(),
1953 converted_arrow_schema.columns().len()
1954 );
1955 parquet_schema
1956 .columns()
1957 .iter()
1958 .zip(converted_arrow_schema.columns())
1959 .for_each(|(a, b)| {
1960 match a.logical_type_ref() {
1965 Some(_) => {
1966 assert_eq!(a, b)
1967 }
1968 None => {
1969 assert_eq!(a.name(), b.name());
1970 assert_eq!(a.physical_type(), b.physical_type());
1971 assert_eq!(a.converted_type(), b.converted_type());
1972 }
1973 }
1974 });
1975 }
1976
1977 #[test]
1978 #[should_panic(expected = "Parquet does not support writing empty structs")]
1979 fn test_empty_struct_field() {
1980 let arrow_fields = vec![Field::new(
1981 "struct",
1982 DataType::Struct(Fields::empty()),
1983 false,
1984 )];
1985 let arrow_schema = Schema::new(arrow_fields);
1986 let converted_arrow_schema = ArrowSchemaConverter::new()
1987 .with_coerce_types(true)
1988 .convert(&arrow_schema);
1989
1990 converted_arrow_schema.unwrap();
1991 }
1992
1993 #[test]
1994 fn test_metadata() {
1995 let message_type = "
1996 message test_schema {
1997 OPTIONAL BINARY string (STRING);
1998 }
1999 ";
2000 let parquet_group_type = parse_message_type(message_type).unwrap();
2001
2002 let key_value_metadata = vec![
2003 KeyValue::new("foo".to_owned(), Some("bar".to_owned())),
2004 KeyValue::new("baz".to_owned(), None),
2005 ];
2006
2007 let mut expected_metadata: HashMap<String, String> = HashMap::new();
2008 expected_metadata.insert("foo".to_owned(), "bar".to_owned());
2009
2010 let parquet_schema = SchemaDescriptor::new(Arc::new(parquet_group_type));
2011 let converted_arrow_schema =
2012 parquet_to_arrow_schema(&parquet_schema, Some(&key_value_metadata)).unwrap();
2013
2014 assert_eq!(converted_arrow_schema.metadata(), &expected_metadata);
2015 }
2016
2017 #[test]
2018 fn test_arrow_schema_roundtrip() -> Result<()> {
2019 let meta = |a: &[(&str, &str)]| -> HashMap<String, String> {
2020 a.iter()
2021 .map(|(a, b)| (a.to_string(), b.to_string()))
2022 .collect()
2023 };
2024
2025 let schema = Schema::new_with_metadata(
2026 vec![
2027 Field::new("c1", DataType::Utf8, false)
2028 .with_metadata(meta(&[("Key", "Foo"), (PARQUET_FIELD_ID_META_KEY, "2")])),
2029 Field::new("c2", DataType::Binary, false),
2030 Field::new("c3", DataType::FixedSizeBinary(3), false),
2031 Field::new("c4", DataType::Boolean, false),
2032 Field::new("c5", DataType::Date32, false),
2033 Field::new("c6", DataType::Date64, false),
2034 Field::new("c7", DataType::Time32(TimeUnit::Second), false),
2035 Field::new("c8", DataType::Time32(TimeUnit::Millisecond), false),
2036 Field::new("c13", DataType::Time64(TimeUnit::Microsecond), false),
2037 Field::new("c14", DataType::Time64(TimeUnit::Nanosecond), false),
2038 Field::new("c15", DataType::Timestamp(TimeUnit::Second, None), false),
2039 Field::new(
2040 "c16",
2041 DataType::Timestamp(TimeUnit::Millisecond, Some("UTC".into())),
2042 false,
2043 ),
2044 Field::new(
2045 "c17",
2046 DataType::Timestamp(TimeUnit::Microsecond, Some("Africa/Johannesburg".into())),
2047 false,
2048 ),
2049 Field::new(
2050 "c18",
2051 DataType::Timestamp(TimeUnit::Nanosecond, None),
2052 false,
2053 ),
2054 Field::new("c19", DataType::Interval(IntervalUnit::DayTime), false),
2055 Field::new("c20", DataType::Interval(IntervalUnit::YearMonth), false),
2056 Field::new_list(
2057 "c21",
2058 Field::new_list_field(DataType::Boolean, true)
2059 .with_metadata(meta(&[("Key", "Bar"), (PARQUET_FIELD_ID_META_KEY, "5")])),
2060 false,
2061 )
2062 .with_metadata(meta(&[(PARQUET_FIELD_ID_META_KEY, "4")])),
2063 Field::new(
2064 "c22",
2065 DataType::FixedSizeList(
2066 Arc::new(Field::new_list_field(DataType::Boolean, true)),
2067 5,
2068 ),
2069 false,
2070 ),
2071 Field::new_list(
2072 "c23",
2073 Field::new_large_list(
2074 "inner",
2075 Field::new_list_field(
2076 DataType::Struct(
2077 vec![
2078 Field::new("a", DataType::Int16, true),
2079 Field::new("b", DataType::Float64, false),
2080 Field::new("c", DataType::Float32, false),
2081 Field::new("d", DataType::Float16, false),
2082 ]
2083 .into(),
2084 ),
2085 false,
2086 ),
2087 true,
2088 ),
2089 false,
2090 ),
2091 Field::new(
2092 "c24",
2093 DataType::Struct(Fields::from(vec![
2094 Field::new("a", DataType::Utf8, false),
2095 Field::new("b", DataType::UInt16, false),
2096 ])),
2097 false,
2098 ),
2099 Field::new("c25", DataType::Interval(IntervalUnit::YearMonth), true),
2100 Field::new("c26", DataType::Interval(IntervalUnit::DayTime), true),
2101 #[expect(deprecated)]
2107 Field::new_dict(
2108 "c31",
2109 DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Utf8)),
2110 true,
2111 123,
2112 true,
2113 )
2114 .with_metadata(meta(&[(PARQUET_FIELD_ID_META_KEY, "6")])),
2115 Field::new("c32", DataType::LargeBinary, true),
2116 Field::new("c33", DataType::LargeUtf8, true),
2117 Field::new_large_list(
2118 "c34",
2119 Field::new_list(
2120 "inner",
2121 Field::new_list_field(
2122 DataType::Struct(
2123 vec![
2124 Field::new("a", DataType::Int16, true),
2125 Field::new("b", DataType::Float64, true),
2126 ]
2127 .into(),
2128 ),
2129 true,
2130 ),
2131 true,
2132 ),
2133 true,
2134 ),
2135 Field::new("c35", DataType::Null, true),
2136 Field::new("c36", DataType::Decimal128(2, 1), false),
2137 Field::new("c37", DataType::Decimal256(50, 20), false),
2138 Field::new("c38", DataType::Decimal128(18, 12), true),
2139 Field::new_map(
2140 "c39",
2141 "key_value",
2142 Field::new("key", DataType::Utf8, false),
2143 Field::new_list("value", Field::new("element", DataType::Utf8, true), true),
2144 false, true,
2146 ),
2147 Field::new_map(
2148 "c40",
2149 "my_entries",
2150 Field::new("my_key", DataType::Utf8, false)
2151 .with_metadata(meta(&[(PARQUET_FIELD_ID_META_KEY, "8")])),
2152 Field::new_list(
2153 "my_value",
2154 Field::new_list_field(DataType::Utf8, true)
2155 .with_metadata(meta(&[(PARQUET_FIELD_ID_META_KEY, "10")])),
2156 true,
2157 )
2158 .with_metadata(meta(&[(PARQUET_FIELD_ID_META_KEY, "9")])),
2159 false, true,
2161 )
2162 .with_metadata(meta(&[(PARQUET_FIELD_ID_META_KEY, "7")])),
2163 Field::new_map(
2164 "c41",
2165 "my_entries",
2166 Field::new("my_key", DataType::Utf8, false),
2167 Field::new_list(
2168 "my_value",
2169 Field::new_list_field(DataType::Utf8, true)
2170 .with_metadata(meta(&[(PARQUET_FIELD_ID_META_KEY, "11")])),
2171 true,
2172 ),
2173 false, false,
2175 ),
2176 Field::new("c42", DataType::Decimal32(5, 2), false),
2177 Field::new("c43", DataType::Decimal64(18, 12), true),
2178 ],
2179 meta(&[("Key", "Value")]),
2180 );
2181
2182 let file = tempfile::tempfile().unwrap();
2184 let writer =
2185 ArrowWriter::try_new(file.try_clone().unwrap(), Arc::new(schema.clone()), None)?;
2186 writer.close()?;
2187
2188 let arrow_reader = ParquetRecordBatchReaderBuilder::try_new(file).unwrap();
2190
2191 let read_schema = arrow_reader.schema();
2193 assert_eq!(&schema, read_schema.as_ref());
2194
2195 let mut stack = Vec::with_capacity(10);
2197 let mut out = Vec::with_capacity(10);
2198
2199 let root = arrow_reader.parquet_schema().root_schema_ptr();
2200 stack.push((root.name().to_string(), root));
2201
2202 while let Some((p, t)) = stack.pop() {
2203 if t.is_group() {
2204 for f in t.get_fields() {
2205 stack.push((format!("{p}.{}", f.name()), f.clone()))
2206 }
2207 }
2208
2209 let info = t.get_basic_info();
2210 if info.has_id() {
2211 out.push(format!("{p} -> {}", info.id()))
2212 }
2213 }
2214 out.sort_unstable();
2215 let out: Vec<_> = out.iter().map(|x| x.as_str()).collect();
2216
2217 assert_eq!(
2218 &out,
2219 &[
2220 "arrow_schema.c1 -> 2",
2221 "arrow_schema.c21 -> 4",
2222 "arrow_schema.c21.list.item -> 5",
2223 "arrow_schema.c31 -> 6",
2224 "arrow_schema.c40 -> 7",
2225 "arrow_schema.c40.my_entries.my_key -> 8",
2226 "arrow_schema.c40.my_entries.my_value -> 9",
2227 "arrow_schema.c40.my_entries.my_value.list.item -> 10",
2228 "arrow_schema.c41.my_entries.my_value.list.item -> 11",
2229 ]
2230 );
2231
2232 Ok(())
2233 }
2234
2235 #[test]
2236 fn test_read_parquet_field_ids_raw() -> Result<()> {
2237 let meta = |a: &[(&str, &str)]| -> HashMap<String, String> {
2238 a.iter()
2239 .map(|(a, b)| (a.to_string(), b.to_string()))
2240 .collect()
2241 };
2242 let schema = Schema::new_with_metadata(
2243 vec![
2244 Field::new("c1", DataType::Utf8, true)
2245 .with_metadata(meta(&[(PARQUET_FIELD_ID_META_KEY, "1")])),
2246 Field::new("c2", DataType::Utf8, true)
2247 .with_metadata(meta(&[(PARQUET_FIELD_ID_META_KEY, "2")])),
2248 ],
2249 HashMap::new(),
2250 );
2251
2252 let writer = ArrowWriter::try_new(vec![], Arc::new(schema.clone()), None)?;
2253 let parquet_bytes = writer.into_inner()?;
2254
2255 let reader =
2256 crate::file::reader::SerializedFileReader::new(bytes::Bytes::from(parquet_bytes))?;
2257 let schema_descriptor = reader.metadata().file_metadata().schema_descr_ptr();
2258
2259 let arrow_schema = crate::arrow::parquet_to_arrow_schema(&schema_descriptor, None)?;
2261
2262 let parq_schema_descr = ArrowSchemaConverter::new()
2263 .with_coerce_types(true)
2264 .convert(&arrow_schema)?;
2265 let parq_fields = parq_schema_descr.root_schema().get_fields();
2266 assert_eq!(parq_fields.len(), 2);
2267 assert_eq!(parq_fields[0].get_basic_info().id(), 1);
2268 assert_eq!(parq_fields[1].get_basic_info().id(), 2);
2269
2270 Ok(())
2271 }
2272
2273 #[test]
2274 fn test_arrow_schema_roundtrip_lists() -> Result<()> {
2275 let metadata = HashMap::from([("Key".to_string(), "Value".to_string())]);
2276
2277 let schema = Schema::new_with_metadata(
2278 vec![
2279 Field::new_list("c21", Field::new("array", DataType::Boolean, true), false),
2280 Field::new(
2281 "c22",
2282 DataType::FixedSizeList(
2283 Arc::new(Field::new("items", DataType::Boolean, false)),
2284 5,
2285 ),
2286 false,
2287 ),
2288 Field::new_list(
2289 "c23",
2290 Field::new_large_list(
2291 "items",
2292 Field::new_struct(
2293 "items",
2294 vec![
2295 Field::new("a", DataType::Int16, true),
2296 Field::new("b", DataType::Float64, false),
2297 ],
2298 true,
2299 ),
2300 true,
2301 ),
2302 true,
2303 ),
2304 ],
2305 metadata,
2306 );
2307
2308 let file = tempfile::tempfile().unwrap();
2310 let writer =
2311 ArrowWriter::try_new(file.try_clone().unwrap(), Arc::new(schema.clone()), None)?;
2312 writer.close()?;
2313
2314 let arrow_reader = ParquetRecordBatchReaderBuilder::try_new(file).unwrap();
2316 let read_schema = arrow_reader.schema();
2317 assert_eq!(&schema, read_schema.as_ref());
2318 Ok(())
2319 }
2320
2321 #[test]
2322 fn test_get_arrow_schema_from_metadata() {
2323 assert!(get_arrow_schema_from_metadata("").is_err());
2324 }
2325
2326 #[test]
2327 #[cfg(feature = "arrow_canonical_extension_types")]
2328 fn arrow_uuid_to_parquet_uuid() -> Result<()> {
2329 use arrow_schema::extension::Uuid;
2330 let arrow_schema = Schema::new(vec![
2331 Field::new("uuid", DataType::FixedSizeBinary(16), false).with_extension_type(Uuid),
2332 ]);
2333
2334 let parquet_schema = ArrowSchemaConverter::new().convert(&arrow_schema)?;
2335
2336 assert_eq!(
2337 parquet_schema.column(0).logical_type_ref(),
2338 Some(&LogicalType::Uuid)
2339 );
2340
2341 let arrow_schema = parquet_to_arrow_schema(&parquet_schema, None)?;
2342 assert_eq!(arrow_schema.field(0).try_extension_type::<Uuid>()?, Uuid);
2343
2344 Ok(())
2345 }
2346
2347 #[test]
2348 #[cfg(feature = "arrow_canonical_extension_types")]
2349 fn arrow_json_to_parquet_json() -> Result<()> {
2350 use arrow_schema::extension::Json;
2351 let arrow_schema = Schema::new(vec![
2352 Field::new("json", DataType::Utf8, false).with_extension_type(Json::default()),
2353 ]);
2354
2355 let parquet_schema = ArrowSchemaConverter::new().convert(&arrow_schema)?;
2356
2357 assert_eq!(
2358 parquet_schema.column(0).logical_type_ref(),
2359 Some(&LogicalType::Json)
2360 );
2361
2362 let arrow_schema = parquet_to_arrow_schema(&parquet_schema, None)?;
2363 assert_eq!(
2364 arrow_schema.field(0).try_extension_type::<Json>()?,
2365 Json::default()
2366 );
2367
2368 Ok(())
2369 }
2370
2371 #[test]
2372 fn test_parquet_to_arrow_field_levels_with_virtual_rejects_non_virtual() {
2373 let message_type = "
2374 message test_schema {
2375 REQUIRED INT32 id;
2376 }
2377 ";
2378 let parquet_schema = Arc::new(parse_message_type(message_type).unwrap());
2379 let descriptor = SchemaDescriptor::new(parquet_schema);
2380
2381 let regular_field = Arc::new(Field::new("regular_column", DataType::Int64, false));
2383 let result = parquet_to_arrow_field_levels_with_virtual(
2384 &descriptor,
2385 ProjectionMask::all(),
2386 None,
2387 &[regular_field],
2388 );
2389
2390 assert!(result.is_err());
2391 assert!(
2392 result
2393 .unwrap_err()
2394 .to_string()
2395 .contains("is not a virtual column")
2396 );
2397 }
2398}