1use std::sync::Arc;
19
20use arrow_array::{
21 Array, ArrayRef, BinaryArray, BooleanArray, Date32Array, Decimal128Array, FixedSizeBinaryArray,
22 FixedSizeListArray, Float32Array, Float64Array, Int32Array, Int64Array, LargeBinaryArray,
23 LargeListArray, LargeStringArray, ListArray, MapArray, StringArray, StructArray,
24 Time64MicrosecondArray, TimestampMicrosecondArray, TimestampNanosecondArray, new_null_array,
25};
26use arrow_buffer::NullBuffer;
27use arrow_schema::{DataType, FieldRef, TimeUnit};
28use uuid::Uuid;
29
30use super::get_field_id_from_metadata;
31use crate::spec::{
32 ListType, Literal, Map, MapType, NestedField, PartnerAccessor, PrimitiveLiteral, PrimitiveType,
33 SchemaWithPartnerVisitor, Struct, StructType, Type, VariantType, visit_struct_with_partner,
34 visit_type_with_partner,
35};
36use crate::{Error, ErrorKind, Result};
37
38struct ArrowArrayToIcebergStructConverter;
39
40impl SchemaWithPartnerVisitor<ArrayRef> for ArrowArrayToIcebergStructConverter {
41 type T = Vec<Option<Literal>>;
42
43 fn schema(
44 &mut self,
45 _schema: &crate::spec::Schema,
46 _partner: &ArrayRef,
47 value: Vec<Option<Literal>>,
48 ) -> Result<Vec<Option<Literal>>> {
49 Ok(value)
50 }
51
52 fn field(
53 &mut self,
54 field: &crate::spec::NestedFieldRef,
55 _partner: &ArrayRef,
56 value: Vec<Option<Literal>>,
57 ) -> Result<Vec<Option<Literal>>> {
58 if field.required && value.iter().any(Option::is_none) {
60 return Err(Error::new(
61 ErrorKind::DataInvalid,
62 "The field is required but has null value",
63 )
64 .with_context("field_id", field.id.to_string())
65 .with_context("field_name", &field.name));
66 }
67 Ok(value)
68 }
69
70 fn r#struct(
71 &mut self,
72 _struct: &StructType,
73 array: &ArrayRef,
74 results: Vec<Vec<Option<Literal>>>,
75 ) -> Result<Vec<Option<Literal>>> {
76 let row_len = results.first().map(|column| column.len()).unwrap_or(0);
77 if let Some(col) = results.iter().find(|col| col.len() != row_len) {
78 return Err(Error::new(
79 ErrorKind::DataInvalid,
80 "The struct columns have different row length",
81 )
82 .with_context("first col length", row_len.to_string())
83 .with_context("actual col length", col.len().to_string()));
84 }
85
86 let mut struct_literals = Vec::with_capacity(row_len);
87 let mut columns_iters = results
88 .into_iter()
89 .map(|column| column.into_iter())
90 .collect::<Vec<_>>();
91
92 for i in 0..row_len {
93 let mut literals = Vec::with_capacity(columns_iters.len());
94 for column_iter in columns_iters.iter_mut() {
95 literals.push(column_iter.next().unwrap());
96 }
97 if array.is_null(i) {
98 struct_literals.push(None);
99 } else {
100 struct_literals.push(Some(Literal::Struct(Struct::from_iter(literals))));
101 }
102 }
103
104 Ok(struct_literals)
105 }
106
107 fn list(
108 &mut self,
109 list: &ListType,
110 array: &ArrayRef,
111 elements: Vec<Option<Literal>>,
112 ) -> Result<Vec<Option<Literal>>> {
113 if list.element_field.required && elements.iter().any(Option::is_none) {
114 return Err(Error::new(
115 ErrorKind::DataInvalid,
116 "The list should not have null value",
117 ));
118 }
119 match array.data_type() {
120 DataType::List(_) => {
121 let offset = array
122 .as_any()
123 .downcast_ref::<ListArray>()
124 .ok_or_else(|| {
125 Error::new(ErrorKind::DataInvalid, "The partner is not a list array")
126 })?
127 .offsets();
128 let mut result = Vec::with_capacity(offset.len() - 1);
130 for i in 0..offset.len() - 1 {
131 let start = offset[i] as usize;
132 let end = offset[i + 1] as usize;
133 result.push(Some(Literal::List(elements[start..end].to_vec())));
134 }
135 Ok(result)
136 }
137 DataType::LargeList(_) => {
138 let offset = array
139 .as_any()
140 .downcast_ref::<LargeListArray>()
141 .ok_or_else(|| {
142 Error::new(
143 ErrorKind::DataInvalid,
144 "The partner is not a large list array",
145 )
146 })?
147 .offsets();
148 let mut result = Vec::with_capacity(offset.len() - 1);
150 for i in 0..offset.len() - 1 {
151 let start = offset[i] as usize;
152 let end = offset[i + 1] as usize;
153 result.push(Some(Literal::List(elements[start..end].to_vec())));
154 }
155 Ok(result)
156 }
157 DataType::FixedSizeList(_, len) => {
158 let mut result = Vec::with_capacity(elements.len() / *len as usize);
159 for i in 0..elements.len() / *len as usize {
160 let start = i * *len as usize;
161 let end = (i + 1) * *len as usize;
162 result.push(Some(Literal::List(elements[start..end].to_vec())));
163 }
164 Ok(result)
165 }
166 _ => Err(Error::new(
167 ErrorKind::DataInvalid,
168 "The partner is not a list type",
169 )),
170 }
171 }
172
173 fn map(
174 &mut self,
175 _map: &MapType,
176 partner: &ArrayRef,
177 key_values: Vec<Option<Literal>>,
178 values: Vec<Option<Literal>>,
179 ) -> Result<Vec<Option<Literal>>> {
180 if key_values.len() != values.len() {
182 return Err(Error::new(
183 ErrorKind::DataInvalid,
184 "The key value and value of map should have the same row length",
185 ));
186 }
187
188 let offsets = partner
189 .as_any()
190 .downcast_ref::<MapArray>()
191 .ok_or_else(|| Error::new(ErrorKind::DataInvalid, "The partner is not a map array"))?
192 .offsets();
193 let mut result = Vec::with_capacity(offsets.len() - 1);
195 for i in 0..offsets.len() - 1 {
196 let start = offsets[i] as usize;
197 let end = offsets[i + 1] as usize;
198 let mut map = Map::new();
199 for (key, value) in key_values[start..end].iter().zip(values[start..end].iter()) {
200 map.insert(key.clone().unwrap(), value.clone());
201 }
202 result.push(Some(Literal::Map(map)));
203 }
204 Ok(result)
205 }
206
207 fn primitive(&mut self, p: &PrimitiveType, partner: &ArrayRef) -> Result<Vec<Option<Literal>>> {
208 match p {
209 PrimitiveType::Boolean => {
210 let array = partner
211 .as_any()
212 .downcast_ref::<BooleanArray>()
213 .ok_or_else(|| {
214 Error::new(ErrorKind::DataInvalid, "The partner is not a boolean array")
215 })?;
216 Ok(array.iter().map(|v| v.map(Literal::bool)).collect())
217 }
218 PrimitiveType::Int => {
219 let array = partner
220 .as_any()
221 .downcast_ref::<Int32Array>()
222 .ok_or_else(|| {
223 Error::new(ErrorKind::DataInvalid, "The partner is not a int32 array")
224 })?;
225 Ok(array.iter().map(|v| v.map(Literal::int)).collect())
226 }
227 PrimitiveType::Long => {
228 let array = partner
229 .as_any()
230 .downcast_ref::<Int64Array>()
231 .ok_or_else(|| {
232 Error::new(ErrorKind::DataInvalid, "The partner is not a int64 array")
233 })?;
234 Ok(array.iter().map(|v| v.map(Literal::long)).collect())
235 }
236 PrimitiveType::Float => {
237 let array = partner
238 .as_any()
239 .downcast_ref::<Float32Array>()
240 .ok_or_else(|| {
241 Error::new(ErrorKind::DataInvalid, "The partner is not a float32 array")
242 })?;
243 Ok(array.iter().map(|v| v.map(Literal::float)).collect())
244 }
245 PrimitiveType::Double => {
246 let array = partner
247 .as_any()
248 .downcast_ref::<Float64Array>()
249 .ok_or_else(|| {
250 Error::new(ErrorKind::DataInvalid, "The partner is not a float64 array")
251 })?;
252 Ok(array.iter().map(|v| v.map(Literal::double)).collect())
253 }
254 PrimitiveType::Decimal { precision, scale } => {
255 let array = partner
256 .as_any()
257 .downcast_ref::<Decimal128Array>()
258 .ok_or_else(|| {
259 Error::new(
260 ErrorKind::DataInvalid,
261 "The partner is not a decimal128 array",
262 )
263 })?;
264 if let DataType::Decimal128(arrow_precision, arrow_scale) = array.data_type()
265 && (*arrow_precision as u32 != *precision || *arrow_scale as u32 != *scale)
266 {
267 return Err(Error::new(
268 ErrorKind::DataInvalid,
269 format!(
270 "The precision or scale ({arrow_precision},{arrow_scale}) of arrow decimal128 array is not compatible with iceberg decimal type ({precision},{scale})"
271 ),
272 ));
273 }
274 Ok(array.iter().map(|v| v.map(Literal::decimal)).collect())
275 }
276 PrimitiveType::Date => {
277 let array = partner
278 .as_any()
279 .downcast_ref::<Date32Array>()
280 .ok_or_else(|| {
281 Error::new(ErrorKind::DataInvalid, "The partner is not a date32 array")
282 })?;
283 Ok(array.iter().map(|v| v.map(Literal::date)).collect())
284 }
285 PrimitiveType::Time => {
286 let array = partner
287 .as_any()
288 .downcast_ref::<Time64MicrosecondArray>()
289 .ok_or_else(|| {
290 Error::new(ErrorKind::DataInvalid, "The partner is not a time64 array")
291 })?;
292 Ok(array.iter().map(|v| v.map(Literal::time)).collect())
293 }
294 PrimitiveType::Timestamp => {
295 let array = partner
296 .as_any()
297 .downcast_ref::<TimestampMicrosecondArray>()
298 .ok_or_else(|| {
299 Error::new(
300 ErrorKind::DataInvalid,
301 "The partner is not a timestamp array",
302 )
303 })?;
304 Ok(array.iter().map(|v| v.map(Literal::timestamp)).collect())
305 }
306 PrimitiveType::Timestamptz => {
307 let array = partner
308 .as_any()
309 .downcast_ref::<TimestampMicrosecondArray>()
310 .ok_or_else(|| {
311 Error::new(
312 ErrorKind::DataInvalid,
313 "The partner is not a timestamptz array",
314 )
315 })?;
316 Ok(array.iter().map(|v| v.map(Literal::timestamptz)).collect())
317 }
318 PrimitiveType::TimestampNs => {
319 let array = partner
320 .as_any()
321 .downcast_ref::<TimestampNanosecondArray>()
322 .ok_or_else(|| {
323 Error::new(
324 ErrorKind::DataInvalid,
325 "The partner is not a timestamp_ns array",
326 )
327 })?;
328 Ok(array
329 .iter()
330 .map(|v| v.map(Literal::timestamp_nano))
331 .collect())
332 }
333 PrimitiveType::TimestamptzNs => {
334 let array = partner
335 .as_any()
336 .downcast_ref::<TimestampNanosecondArray>()
337 .ok_or_else(|| {
338 Error::new(
339 ErrorKind::DataInvalid,
340 "The partner is not a timestamptz_ns array",
341 )
342 })?;
343 Ok(array
344 .iter()
345 .map(|v| v.map(Literal::timestamptz_nano))
346 .collect())
347 }
348 PrimitiveType::String => {
349 if let Some(array) = partner.as_any().downcast_ref::<LargeStringArray>() {
350 Ok(array.iter().map(|v| v.map(Literal::string)).collect())
351 } else if let Some(array) = partner.as_any().downcast_ref::<StringArray>() {
352 Ok(array.iter().map(|v| v.map(Literal::string)).collect())
353 } else {
354 Err(Error::new(
355 ErrorKind::DataInvalid,
356 "The partner is not a string array",
357 ))
358 }
359 }
360 PrimitiveType::Uuid => {
361 if let Some(array) = partner.as_any().downcast_ref::<FixedSizeBinaryArray>() {
362 if array.value_length() != 16 {
363 return Err(Error::new(
364 ErrorKind::DataInvalid,
365 "The partner is not a uuid array",
366 ));
367 }
368 Ok(array
369 .iter()
370 .map(|v| {
371 v.map(|v| {
372 Ok(Literal::uuid(Uuid::from_bytes(v.try_into().map_err(
373 |_| {
374 Error::new(
375 ErrorKind::DataInvalid,
376 "Failed to convert binary to uuid",
377 )
378 },
379 )?)))
380 })
381 .transpose()
382 })
383 .collect::<Result<Vec<_>>>()?)
384 } else {
385 Err(Error::new(
386 ErrorKind::DataInvalid,
387 "The partner is not a uuid array",
388 ))
389 }
390 }
391 PrimitiveType::Fixed(len) => {
392 let array = partner
393 .as_any()
394 .downcast_ref::<FixedSizeBinaryArray>()
395 .ok_or_else(|| {
396 Error::new(ErrorKind::DataInvalid, "The partner is not a fixed array")
397 })?;
398 if array.value_length() != *len as i32 {
399 return Err(Error::new(
400 ErrorKind::DataInvalid,
401 "The length of fixed size binary array is not compatible with iceberg fixed type",
402 ));
403 }
404 Ok(array
405 .iter()
406 .map(|v| v.map(|v| Literal::fixed(v.iter().cloned())))
407 .collect())
408 }
409 PrimitiveType::Binary => {
410 if let Some(array) = partner.as_any().downcast_ref::<LargeBinaryArray>() {
411 Ok(array
412 .iter()
413 .map(|v| v.map(|v| Literal::binary(v.to_vec())))
414 .collect())
415 } else if let Some(array) = partner.as_any().downcast_ref::<BinaryArray>() {
416 Ok(array
417 .iter()
418 .map(|v| v.map(|v| Literal::binary(v.to_vec())))
419 .collect())
420 } else {
421 Err(Error::new(
422 ErrorKind::DataInvalid,
423 "The partner is not a binary array",
424 ))
425 }
426 }
427 }
428 }
429
430 fn variant(&mut self, _v: &VariantType, _partner: &ArrayRef) -> Result<Vec<Option<Literal>>> {
431 Err(Error::new(
432 ErrorKind::FeatureUnsupported,
433 "Converting variant Arrow array to Iceberg literal is not supported yet",
434 ))
435 }
436}
437
438#[derive(Clone, Copy, Debug)]
449pub enum FieldMatchMode {
450 Id,
452 Name,
454}
455
456impl FieldMatchMode {
457 pub fn match_field(&self, arrow_field: &FieldRef, iceberg_field: &NestedField) -> bool {
459 match self {
460 FieldMatchMode::Id => get_field_id_from_metadata(arrow_field)
461 .map(|id| id == iceberg_field.id)
462 .unwrap_or(false),
463 FieldMatchMode::Name => arrow_field.name() == &iceberg_field.name,
464 }
465 }
466}
467
468pub struct ArrowArrayAccessor {
470 match_mode: FieldMatchMode,
471}
472
473impl ArrowArrayAccessor {
474 pub fn new() -> Self {
476 Self {
477 match_mode: FieldMatchMode::Id,
478 }
479 }
480
481 pub fn new_with_match_mode(match_mode: FieldMatchMode) -> Self {
483 Self { match_mode }
484 }
485}
486
487impl Default for ArrowArrayAccessor {
488 fn default() -> Self {
489 Self::new()
490 }
491}
492
493impl PartnerAccessor<ArrayRef> for ArrowArrayAccessor {
494 fn struct_partner<'a>(&self, schema_partner: &'a ArrayRef) -> Result<&'a ArrayRef> {
495 if !matches!(schema_partner.data_type(), DataType::Struct(_)) {
496 return Err(Error::new(
497 ErrorKind::DataInvalid,
498 "The schema partner is not a struct type",
499 ));
500 }
501
502 Ok(schema_partner)
503 }
504
505 fn field_partner<'a>(
506 &self,
507 struct_partner: &'a ArrayRef,
508 field: &NestedField,
509 ) -> Result<&'a ArrayRef> {
510 let struct_array = struct_partner
511 .as_any()
512 .downcast_ref::<StructArray>()
513 .ok_or_else(|| {
514 Error::new(
515 ErrorKind::DataInvalid,
516 format!(
517 "The struct partner is not a struct array, partner: {struct_partner:?}"
518 ),
519 )
520 })?;
521
522 let field_pos = struct_array
523 .fields()
524 .iter()
525 .position(|arrow_field| self.match_mode.match_field(arrow_field, field))
526 .ok_or_else(|| {
527 Error::new(
528 ErrorKind::DataInvalid,
529 format!("Field id {} not found in struct array", field.id),
530 )
531 })?;
532
533 Ok(struct_array.column(field_pos))
534 }
535
536 fn list_element_partner<'a>(&self, list_partner: &'a ArrayRef) -> Result<&'a ArrayRef> {
537 match list_partner.data_type() {
538 DataType::List(_) => {
539 let list_array = list_partner
540 .as_any()
541 .downcast_ref::<ListArray>()
542 .ok_or_else(|| {
543 Error::new(
544 ErrorKind::DataInvalid,
545 "The list partner is not a list array",
546 )
547 })?;
548 Ok(list_array.values())
549 }
550 DataType::LargeList(_) => {
551 let list_array = list_partner
552 .as_any()
553 .downcast_ref::<LargeListArray>()
554 .ok_or_else(|| {
555 Error::new(
556 ErrorKind::DataInvalid,
557 "The list partner is not a large list array",
558 )
559 })?;
560 Ok(list_array.values())
561 }
562 DataType::FixedSizeList(_, _) => {
563 let list_array = list_partner
564 .as_any()
565 .downcast_ref::<FixedSizeListArray>()
566 .ok_or_else(|| {
567 Error::new(
568 ErrorKind::DataInvalid,
569 "The list partner is not a fixed size list array",
570 )
571 })?;
572 Ok(list_array.values())
573 }
574 _ => Err(Error::new(
575 ErrorKind::DataInvalid,
576 "The list partner is not a list type",
577 )),
578 }
579 }
580
581 fn map_key_partner<'a>(&self, map_partner: &'a ArrayRef) -> Result<&'a ArrayRef> {
582 let map_array = map_partner
583 .as_any()
584 .downcast_ref::<MapArray>()
585 .ok_or_else(|| {
586 Error::new(ErrorKind::DataInvalid, "The map partner is not a map array")
587 })?;
588 Ok(map_array.keys())
589 }
590
591 fn map_value_partner<'a>(&self, map_partner: &'a ArrayRef) -> Result<&'a ArrayRef> {
592 let map_array = map_partner
593 .as_any()
594 .downcast_ref::<MapArray>()
595 .ok_or_else(|| {
596 Error::new(ErrorKind::DataInvalid, "The map partner is not a map array")
597 })?;
598 Ok(map_array.values())
599 }
600}
601
602pub fn arrow_struct_to_literal(
605 struct_array: &ArrayRef,
606 ty: &StructType,
607) -> Result<Vec<Option<Literal>>> {
608 visit_struct_with_partner(
609 ty,
610 struct_array,
611 &mut ArrowArrayToIcebergStructConverter,
612 &ArrowArrayAccessor::new(),
613 )
614}
615
616pub fn arrow_primitive_to_literal(
619 primitive_array: &ArrayRef,
620 ty: &Type,
621) -> Result<Vec<Option<Literal>>> {
622 visit_type_with_partner(
623 ty,
624 primitive_array,
625 &mut ArrowArrayToIcebergStructConverter,
626 &ArrowArrayAccessor::new(),
627 )
628}
629
630pub(crate) fn create_primitive_array_single_element(
635 data_type: &DataType,
636 prim_lit: &Option<PrimitiveLiteral>,
637) -> Result<ArrayRef> {
638 match (data_type, prim_lit) {
639 (DataType::Boolean, Some(PrimitiveLiteral::Boolean(v))) => {
640 Ok(Arc::new(BooleanArray::from(vec![*v])))
641 }
642 (DataType::Boolean, None) => Ok(Arc::new(BooleanArray::from(vec![Option::<bool>::None]))),
643 (DataType::Int32, Some(PrimitiveLiteral::Int(v))) => {
644 Ok(Arc::new(Int32Array::from(vec![*v])))
645 }
646 (DataType::Int32, None) => Ok(Arc::new(Int32Array::from(vec![Option::<i32>::None]))),
647 (DataType::Date32, Some(PrimitiveLiteral::Int(v))) => {
648 Ok(Arc::new(Date32Array::from(vec![*v])))
649 }
650 (DataType::Date32, None) => Ok(Arc::new(Date32Array::from(vec![Option::<i32>::None]))),
651 (DataType::Int64, Some(PrimitiveLiteral::Long(v))) => {
652 Ok(Arc::new(Int64Array::from(vec![*v])))
653 }
654 (DataType::Int64, None) => Ok(Arc::new(Int64Array::from(vec![Option::<i64>::None]))),
655 (DataType::Timestamp(TimeUnit::Microsecond, timezone), Some(PrimitiveLiteral::Long(v))) => {
656 let array = TimestampMicrosecondArray::from(vec![*v]);
657 if let Some(timezone) = timezone {
658 Ok(Arc::new(array.with_timezone(timezone.clone())))
659 } else {
660 Ok(Arc::new(array))
661 }
662 }
663 (DataType::Timestamp(TimeUnit::Microsecond, timezone), None) => {
664 let array = TimestampMicrosecondArray::from(vec![Option::<i64>::None]);
665 if let Some(timezone) = timezone {
666 Ok(Arc::new(array.with_timezone(timezone.clone())))
667 } else {
668 Ok(Arc::new(array))
669 }
670 }
671 (DataType::Timestamp(TimeUnit::Nanosecond, timezone), Some(PrimitiveLiteral::Long(v))) => {
672 let array = TimestampNanosecondArray::from(vec![*v]);
673 if let Some(timezone) = timezone {
674 Ok(Arc::new(array.with_timezone(timezone.clone())))
675 } else {
676 Ok(Arc::new(array))
677 }
678 }
679 (DataType::Timestamp(TimeUnit::Nanosecond, timezone), None) => {
680 let array = TimestampNanosecondArray::from(vec![Option::<i64>::None]);
681 if let Some(timezone) = timezone {
682 Ok(Arc::new(array.with_timezone(timezone.clone())))
683 } else {
684 Ok(Arc::new(array))
685 }
686 }
687 (DataType::Float32, Some(PrimitiveLiteral::Float(v))) => {
688 Ok(Arc::new(Float32Array::from(vec![v.0])))
689 }
690 (DataType::Float32, None) => Ok(Arc::new(Float32Array::from(vec![Option::<f32>::None]))),
691 (DataType::Float64, Some(PrimitiveLiteral::Double(v))) => {
692 Ok(Arc::new(Float64Array::from(vec![v.0])))
693 }
694 (DataType::Float64, None) => Ok(Arc::new(Float64Array::from(vec![Option::<f64>::None]))),
695 (DataType::Utf8, Some(PrimitiveLiteral::String(v))) => {
696 Ok(Arc::new(StringArray::from(vec![v.as_str()])))
697 }
698 (DataType::Utf8, None) => Ok(Arc::new(StringArray::from(vec![Option::<&str>::None]))),
699 (DataType::Binary, Some(PrimitiveLiteral::Binary(v))) => {
700 Ok(Arc::new(BinaryArray::from_vec(vec![v.as_slice()])))
701 }
702 (DataType::Binary, None) => Ok(Arc::new(BinaryArray::from_opt_vec(vec![
703 Option::<&[u8]>::None,
704 ]))),
705 (DataType::Decimal128(precision, scale), Some(PrimitiveLiteral::Int128(v))) => {
706 let array = Decimal128Array::from(vec![{ *v }])
707 .with_precision_and_scale(*precision, *scale)
708 .map_err(|e| {
709 Error::new(
710 ErrorKind::DataInvalid,
711 format!(
712 "Failed to create Decimal128Array with precision {precision} and scale {scale}: {e}"
713 ),
714 )
715 })?;
716 Ok(Arc::new(array))
717 }
718 (DataType::Decimal128(precision, scale), Some(PrimitiveLiteral::UInt128(v))) => {
719 let array = Decimal128Array::from(vec![*v as i128])
720 .with_precision_and_scale(*precision, *scale)
721 .map_err(|e| {
722 Error::new(
723 ErrorKind::DataInvalid,
724 format!(
725 "Failed to create Decimal128Array with precision {precision} and scale {scale}: {e}"
726 ),
727 )
728 })?;
729 Ok(Arc::new(array))
730 }
731 (DataType::Decimal128(precision, scale), None) => {
732 let array = Decimal128Array::from(vec![Option::<i128>::None])
733 .with_precision_and_scale(*precision, *scale)
734 .map_err(|e| {
735 Error::new(
736 ErrorKind::DataInvalid,
737 format!(
738 "Failed to create Decimal128Array with precision {precision} and scale {scale}: {e}"
739 ),
740 )
741 })?;
742 Ok(Arc::new(array))
743 }
744 (DataType::Struct(fields), None) => {
745 let null_arrays: Vec<ArrayRef> = fields
747 .iter()
748 .map(|f| {
749 match f.data_type() {
752 DataType::Boolean => {
753 Ok(Arc::new(BooleanArray::from(vec![Option::<bool>::None]))
754 as ArrayRef)
755 }
756 DataType::Int32 | DataType::Date32 => {
757 Ok(Arc::new(Int32Array::from(vec![Option::<i32>::None])) as ArrayRef)
758 }
759 DataType::Int64 => {
760 Ok(Arc::new(Int64Array::from(vec![Option::<i64>::None])) as ArrayRef)
761 }
762 DataType::Timestamp(TimeUnit::Microsecond, timezone) => {
763 let array = TimestampMicrosecondArray::from(vec![Option::<i64>::None]);
764 if let Some(timezone) = timezone {
765 Ok(Arc::new(array.with_timezone(timezone.clone())) as ArrayRef)
766 } else {
767 Ok(Arc::new(array) as ArrayRef)
768 }
769 }
770 DataType::Timestamp(TimeUnit::Nanosecond, timezone) => {
771 let array = TimestampNanosecondArray::from(vec![Option::<i64>::None]);
772 if let Some(timezone) = timezone {
773 Ok(Arc::new(array.with_timezone(timezone.clone())) as ArrayRef)
774 } else {
775 Ok(Arc::new(array) as ArrayRef)
776 }
777 }
778 DataType::Float32 => {
779 Ok(Arc::new(Float32Array::from(vec![Option::<f32>::None])) as ArrayRef)
780 }
781 DataType::Float64 => {
782 Ok(Arc::new(Float64Array::from(vec![Option::<f64>::None])) as ArrayRef)
783 }
784 DataType::Utf8 => {
785 Ok(Arc::new(StringArray::from(vec![Option::<&str>::None])) as ArrayRef)
786 }
787 DataType::Binary => {
788 Ok(
789 Arc::new(BinaryArray::from_opt_vec(vec![Option::<&[u8]>::None]))
790 as ArrayRef,
791 )
792 }
793 _ => Err(Error::new(
794 ErrorKind::Unexpected,
795 format!("Unsupported struct field type: {:?}", f.data_type()),
796 )),
797 }
798 })
799 .collect::<Result<Vec<_>>>()?;
800 Ok(Arc::new(StructArray::new(
801 fields.clone(),
802 null_arrays,
803 Some(NullBuffer::new_null(1)),
804 )))
805 }
806 _ => Err(Error::new(
807 ErrorKind::Unexpected,
808 format!("Unsupported constant type combination: {data_type:?} with {prim_lit:?}"),
809 )),
810 }
811}
812
813pub(crate) fn create_primitive_array_repeated(
818 data_type: &DataType,
819 prim_lit: &Option<PrimitiveLiteral>,
820 num_rows: usize,
821) -> Result<ArrayRef> {
822 Ok(match (data_type, prim_lit) {
823 (DataType::Boolean, Some(PrimitiveLiteral::Boolean(value))) => {
825 Arc::new(BooleanArray::from(vec![*value; num_rows]))
826 }
827 (DataType::Int32, Some(PrimitiveLiteral::Int(value))) => {
828 Arc::new(Int32Array::from(vec![*value; num_rows]))
829 }
830 (DataType::Date32, Some(PrimitiveLiteral::Int(value))) => {
831 Arc::new(Date32Array::from(vec![*value; num_rows]))
832 }
833 (DataType::Int64, Some(PrimitiveLiteral::Int(value))) => {
834 Arc::new(Int64Array::from(vec![i64::from(*value); num_rows]))
835 }
836 (DataType::Int64, Some(PrimitiveLiteral::Long(value))) => {
837 Arc::new(Int64Array::from(vec![*value; num_rows]))
838 }
839 (
840 DataType::Timestamp(TimeUnit::Microsecond, timezone),
841 Some(PrimitiveLiteral::Long(value)),
842 ) => {
843 let array = TimestampMicrosecondArray::from(vec![*value; num_rows]);
844 if let Some(timezone) = timezone {
845 Arc::new(array.with_timezone(timezone.clone()))
846 } else {
847 Arc::new(array)
848 }
849 }
850 (
851 DataType::Timestamp(TimeUnit::Nanosecond, timezone),
852 Some(PrimitiveLiteral::Long(value)),
853 ) => {
854 let array = TimestampNanosecondArray::from(vec![*value; num_rows]);
855 if let Some(timezone) = timezone {
856 Arc::new(array.with_timezone(timezone.clone()))
857 } else {
858 Arc::new(array)
859 }
860 }
861 (DataType::Float32, Some(PrimitiveLiteral::Float(value))) => {
862 Arc::new(Float32Array::from(vec![value.0; num_rows]))
863 }
864 (DataType::Float64, Some(PrimitiveLiteral::Double(value))) => {
865 Arc::new(Float64Array::from(vec![value.0; num_rows]))
866 }
867 (DataType::Utf8, Some(PrimitiveLiteral::String(value))) => {
868 Arc::new(StringArray::from(vec![value.clone(); num_rows]))
869 }
870 (DataType::Binary, Some(PrimitiveLiteral::Binary(value))) => {
871 Arc::new(BinaryArray::from_vec(vec![value; num_rows]))
872 }
873 (DataType::LargeBinary, Some(PrimitiveLiteral::Binary(value))) => {
874 Arc::new(LargeBinaryArray::from_vec(vec![value; num_rows]))
875 }
876 (DataType::FixedSizeBinary(len), Some(PrimitiveLiteral::Binary(value))) => {
877 let repeated: Vec<&[u8]> = vec![value.as_slice(); num_rows];
878 Arc::new(FixedSizeBinaryArray::try_from_iter(repeated.into_iter()).map_err(|e| {
879 Error::new(
880 ErrorKind::DataInvalid,
881 format!("Failed to create FixedSizeBinary({len}) array: {e}"),
882 )
883 })?)
884 }
885 (DataType::Time64(TimeUnit::Microsecond), Some(PrimitiveLiteral::Long(value))) => {
886 Arc::new(Time64MicrosecondArray::from(vec![*value; num_rows]))
887 }
888 (DataType::Decimal128(precision, scale), Some(PrimitiveLiteral::Int128(value))) => {
889 Arc::new(
890 Decimal128Array::from(vec![*value; num_rows])
891 .with_precision_and_scale(*precision, *scale)
892 .map_err(|e| {
893 Error::new(
894 ErrorKind::DataInvalid,
895 format!(
896 "Failed to create Decimal128Array with precision {precision} and scale {scale}: {e}"
897 ),
898 )
899 })?,
900 )
901 }
902 (DataType::Decimal128(precision, scale), Some(PrimitiveLiteral::UInt128(value))) => {
903 Arc::new(
904 Decimal128Array::from(vec![*value as i128; num_rows])
905 .with_precision_and_scale(*precision, *scale)
906 .map_err(|e| {
907 Error::new(
908 ErrorKind::DataInvalid,
909 format!(
910 "Failed to create Decimal128Array with precision {precision} and scale {scale}: {e}"
911 ),
912 )
913 })?,
914 )
915 }
916
917 (DataType::Decimal128(precision, scale), None) => {
919 let vals: Vec<Option<i128>> = vec![None; num_rows];
920 Arc::new(
921 Decimal128Array::from(vals)
922 .with_precision_and_scale(*precision, *scale)
923 .map_err(|e| {
924 Error::new(
925 ErrorKind::DataInvalid,
926 format!(
927 "Failed to create Decimal128Array with precision {precision} and scale {scale}: {e}"
928 ),
929 )
930 })?,
931 )
932 }
933 (DataType::Struct(fields), None) => {
934 let null_arrays: Vec<ArrayRef> = fields
936 .iter()
937 .map(|field| create_primitive_array_repeated(field.data_type(), &None, num_rows))
938 .collect::<Result<Vec<_>>>()?;
939
940 Arc::new(StructArray::new(
941 fields.clone(),
942 null_arrays,
943 Some(NullBuffer::new_null(num_rows)),
944 ))
945 }
946 (DataType::Null, _) => Arc::new(arrow_array::NullArray::new(num_rows)),
947
948 (dt, None) => new_null_array(dt, num_rows),
950
951 (dt, _) => {
952 return Err(Error::new(
953 ErrorKind::Unexpected,
954 format!("unexpected target column type {dt}, prim_lit {prim_lit:?}"),
955 ));
956 }
957 })
958}
959
960#[cfg(test)]
961mod test {
962 use std::collections::HashMap;
963 use std::sync::Arc;
964
965 use arrow_array::builder::{Int32Builder, ListBuilder, MapBuilder, StructBuilder};
966 use arrow_array::{
967 ArrayRef, BinaryArray, BooleanArray, Date32Array, Decimal128Array, Float32Array,
968 Float64Array, Int32Array, Int64Array, StringArray, StructArray, Time64MicrosecondArray,
969 TimestampMicrosecondArray, TimestampNanosecondArray,
970 };
971 use arrow_schema::{DataType, Field, Fields, TimeUnit};
972 use parquet::arrow::PARQUET_FIELD_ID_META_KEY;
973
974 use super::*;
975 use crate::spec::{ListType, Literal, MapType, NestedField, PrimitiveType, StructType, Type};
976
977 #[test]
978 fn test_arrow_struct_to_iceberg_struct() {
979 let bool_array = BooleanArray::from(vec![Some(true), Some(false), None]);
980 let int32_array = Int32Array::from(vec![Some(3), Some(4), None]);
981 let int64_array = Int64Array::from(vec![Some(5), Some(6), None]);
982 let float32_array = Float32Array::from(vec![Some(1.1), Some(2.2), None]);
983 let float64_array = Float64Array::from(vec![Some(3.3), Some(4.4), None]);
984 let decimal_array = Decimal128Array::from(vec![Some(1000), Some(2000), None])
985 .with_precision_and_scale(10, 2)
986 .unwrap();
987 let date_array = Date32Array::from(vec![Some(18628), Some(18629), None]);
988 let time_array = Time64MicrosecondArray::from(vec![Some(123456789), Some(987654321), None]);
989 let timestamp_micro_array = TimestampMicrosecondArray::from(vec![
990 Some(1622548800000000),
991 Some(1622635200000000),
992 None,
993 ]);
994 let timestamp_nano_array = TimestampNanosecondArray::from(vec![
995 Some(1622548800000000000),
996 Some(1622635200000000000),
997 None,
998 ]);
999 let string_array = StringArray::from(vec![Some("a"), Some("b"), None]);
1000 let binary_array =
1001 BinaryArray::from(vec![Some(b"abc".as_ref()), Some(b"def".as_ref()), None]);
1002
1003 let struct_array = Arc::new(StructArray::from(vec![
1004 (
1005 Arc::new(
1006 Field::new("bool_field", DataType::Boolean, true).with_metadata(HashMap::from(
1007 [(PARQUET_FIELD_ID_META_KEY.to_string(), "0".to_string())],
1008 )),
1009 ),
1010 Arc::new(bool_array) as ArrayRef,
1011 ),
1012 (
1013 Arc::new(
1014 Field::new("int32_field", DataType::Int32, true).with_metadata(HashMap::from(
1015 [(PARQUET_FIELD_ID_META_KEY.to_string(), "2".to_string())],
1016 )),
1017 ),
1018 Arc::new(int32_array) as ArrayRef,
1019 ),
1020 (
1021 Arc::new(
1022 Field::new("int64_field", DataType::Int64, true).with_metadata(HashMap::from(
1023 [(PARQUET_FIELD_ID_META_KEY.to_string(), "3".to_string())],
1024 )),
1025 ),
1026 Arc::new(int64_array) as ArrayRef,
1027 ),
1028 (
1029 Arc::new(
1030 Field::new("float32_field", DataType::Float32, true).with_metadata(
1031 HashMap::from([(PARQUET_FIELD_ID_META_KEY.to_string(), "4".to_string())]),
1032 ),
1033 ),
1034 Arc::new(float32_array) as ArrayRef,
1035 ),
1036 (
1037 Arc::new(
1038 Field::new("float64_field", DataType::Float64, true).with_metadata(
1039 HashMap::from([(PARQUET_FIELD_ID_META_KEY.to_string(), "5".to_string())]),
1040 ),
1041 ),
1042 Arc::new(float64_array) as ArrayRef,
1043 ),
1044 (
1045 Arc::new(
1046 Field::new("decimal_field", DataType::Decimal128(10, 2), true).with_metadata(
1047 HashMap::from([(PARQUET_FIELD_ID_META_KEY.to_string(), "6".to_string())]),
1048 ),
1049 ),
1050 Arc::new(decimal_array) as ArrayRef,
1051 ),
1052 (
1053 Arc::new(
1054 Field::new("date_field", DataType::Date32, true).with_metadata(HashMap::from(
1055 [(PARQUET_FIELD_ID_META_KEY.to_string(), "7".to_string())],
1056 )),
1057 ),
1058 Arc::new(date_array) as ArrayRef,
1059 ),
1060 (
1061 Arc::new(
1062 Field::new("time_field", DataType::Time64(TimeUnit::Microsecond), true)
1063 .with_metadata(HashMap::from([(
1064 PARQUET_FIELD_ID_META_KEY.to_string(),
1065 "8".to_string(),
1066 )])),
1067 ),
1068 Arc::new(time_array) as ArrayRef,
1069 ),
1070 (
1071 Arc::new(
1072 Field::new(
1073 "timestamp_micro_field",
1074 DataType::Timestamp(TimeUnit::Microsecond, None),
1075 true,
1076 )
1077 .with_metadata(HashMap::from([(
1078 PARQUET_FIELD_ID_META_KEY.to_string(),
1079 "9".to_string(),
1080 )])),
1081 ),
1082 Arc::new(timestamp_micro_array) as ArrayRef,
1083 ),
1084 (
1085 Arc::new(
1086 Field::new(
1087 "timestamp_nano_field",
1088 DataType::Timestamp(TimeUnit::Nanosecond, None),
1089 true,
1090 )
1091 .with_metadata(HashMap::from([(
1092 PARQUET_FIELD_ID_META_KEY.to_string(),
1093 "10".to_string(),
1094 )])),
1095 ),
1096 Arc::new(timestamp_nano_array) as ArrayRef,
1097 ),
1098 (
1099 Arc::new(
1100 Field::new("string_field", DataType::Utf8, true).with_metadata(HashMap::from(
1101 [(PARQUET_FIELD_ID_META_KEY.to_string(), "11".to_string())],
1102 )),
1103 ),
1104 Arc::new(string_array) as ArrayRef,
1105 ),
1106 (
1107 Arc::new(
1108 Field::new("binary_field", DataType::Binary, true).with_metadata(
1109 HashMap::from([(PARQUET_FIELD_ID_META_KEY.to_string(), "12".to_string())]),
1110 ),
1111 ),
1112 Arc::new(binary_array) as ArrayRef,
1113 ),
1114 ])) as ArrayRef;
1115
1116 let iceberg_struct_type = StructType::new(vec![
1117 Arc::new(NestedField::optional(
1118 0,
1119 "bool_field",
1120 Type::Primitive(PrimitiveType::Boolean),
1121 )),
1122 Arc::new(NestedField::optional(
1123 2,
1124 "int32_field",
1125 Type::Primitive(PrimitiveType::Int),
1126 )),
1127 Arc::new(NestedField::optional(
1128 3,
1129 "int64_field",
1130 Type::Primitive(PrimitiveType::Long),
1131 )),
1132 Arc::new(NestedField::optional(
1133 4,
1134 "float32_field",
1135 Type::Primitive(PrimitiveType::Float),
1136 )),
1137 Arc::new(NestedField::optional(
1138 5,
1139 "float64_field",
1140 Type::Primitive(PrimitiveType::Double),
1141 )),
1142 Arc::new(NestedField::optional(
1143 6,
1144 "decimal_field",
1145 Type::Primitive(PrimitiveType::Decimal {
1146 precision: 10,
1147 scale: 2,
1148 }),
1149 )),
1150 Arc::new(NestedField::optional(
1151 7,
1152 "date_field",
1153 Type::Primitive(PrimitiveType::Date),
1154 )),
1155 Arc::new(NestedField::optional(
1156 8,
1157 "time_field",
1158 Type::Primitive(PrimitiveType::Time),
1159 )),
1160 Arc::new(NestedField::optional(
1161 9,
1162 "timestamp_micro_field",
1163 Type::Primitive(PrimitiveType::Timestamp),
1164 )),
1165 Arc::new(NestedField::optional(
1166 10,
1167 "timestamp_nao_field",
1168 Type::Primitive(PrimitiveType::TimestampNs),
1169 )),
1170 Arc::new(NestedField::optional(
1171 11,
1172 "string_field",
1173 Type::Primitive(PrimitiveType::String),
1174 )),
1175 Arc::new(NestedField::optional(
1176 12,
1177 "binary_field",
1178 Type::Primitive(PrimitiveType::Binary),
1179 )),
1180 ]);
1181
1182 let result = arrow_struct_to_literal(&struct_array, &iceberg_struct_type).unwrap();
1183
1184 assert_eq!(result, vec![
1185 Some(Literal::Struct(Struct::from_iter(vec![
1186 Some(Literal::bool(true)),
1187 Some(Literal::int(3)),
1188 Some(Literal::long(5)),
1189 Some(Literal::float(1.1)),
1190 Some(Literal::double(3.3)),
1191 Some(Literal::decimal(1000)),
1192 Some(Literal::date(18628)),
1193 Some(Literal::time(123456789)),
1194 Some(Literal::timestamp(1622548800000000)),
1195 Some(Literal::timestamp_nano(1622548800000000000)),
1196 Some(Literal::string("a".to_string())),
1197 Some(Literal::binary(b"abc".to_vec())),
1198 ]))),
1199 Some(Literal::Struct(Struct::from_iter(vec![
1200 Some(Literal::bool(false)),
1201 Some(Literal::int(4)),
1202 Some(Literal::long(6)),
1203 Some(Literal::float(2.2)),
1204 Some(Literal::double(4.4)),
1205 Some(Literal::decimal(2000)),
1206 Some(Literal::date(18629)),
1207 Some(Literal::time(987654321)),
1208 Some(Literal::timestamp(1622635200000000)),
1209 Some(Literal::timestamp_nano(1622635200000000000)),
1210 Some(Literal::string("b".to_string())),
1211 Some(Literal::binary(b"def".to_vec())),
1212 ]))),
1213 Some(Literal::Struct(Struct::from_iter(vec![
1214 None, None, None, None, None, None, None, None, None, None, None, None,
1215 ]))),
1216 ]);
1217 }
1218
1219 #[test]
1220 fn test_nullable_struct() {
1221 let struct_array = {
1228 let mut builder = StructBuilder::from_fields(
1229 Fields::from(vec![
1230 Field::new("a", DataType::Int32, true).with_metadata(HashMap::from([(
1231 PARQUET_FIELD_ID_META_KEY.to_string(),
1232 "0".to_string(),
1233 )])),
1234 Field::new("b", DataType::Int32, true).with_metadata(HashMap::from([(
1235 PARQUET_FIELD_ID_META_KEY.to_string(),
1236 "1".to_string(),
1237 )])),
1238 ]),
1239 3,
1240 );
1241 builder
1242 .field_builder::<Int32Builder>(0)
1243 .unwrap()
1244 .append_null();
1245 builder
1246 .field_builder::<Int32Builder>(1)
1247 .unwrap()
1248 .append_null();
1249 builder.append(true);
1250
1251 builder
1252 .field_builder::<Int32Builder>(0)
1253 .unwrap()
1254 .append_value(1);
1255 builder
1256 .field_builder::<Int32Builder>(1)
1257 .unwrap()
1258 .append_null();
1259 builder.append(true);
1260
1261 builder
1262 .field_builder::<Int32Builder>(0)
1263 .unwrap()
1264 .append_value(1);
1265 builder
1266 .field_builder::<Int32Builder>(1)
1267 .unwrap()
1268 .append_value(1);
1269 builder.append_null();
1270
1271 Arc::new(builder.finish()) as ArrayRef
1272 };
1273
1274 let iceberg_struct_type = StructType::new(vec![
1275 Arc::new(NestedField::optional(
1276 0,
1277 "a",
1278 Type::Primitive(PrimitiveType::Int),
1279 )),
1280 Arc::new(NestedField::optional(
1281 1,
1282 "b",
1283 Type::Primitive(PrimitiveType::Int),
1284 )),
1285 ]);
1286
1287 let result = arrow_struct_to_literal(&struct_array, &iceberg_struct_type).unwrap();
1288 assert_eq!(result, vec![
1289 Some(Literal::Struct(Struct::from_iter(vec![None, None,]))),
1290 Some(Literal::Struct(Struct::from_iter(vec![
1291 Some(Literal::int(1)),
1292 None,
1293 ]))),
1294 None,
1295 ]);
1296 }
1297
1298 #[test]
1299 fn test_empty_struct() {
1300 let struct_array = Arc::new(StructArray::new_null(Fields::empty(), 3)) as ArrayRef;
1301 let iceberg_struct_type = StructType::new(vec![]);
1302 let result = arrow_struct_to_literal(&struct_array, &iceberg_struct_type).unwrap();
1303 assert_eq!(result, vec![None; 0]);
1304 }
1305
1306 #[test]
1307 fn test_arrow_variant_to_literal_is_unsupported() {
1308 let variant_child = Arc::new(StructArray::from(vec![
1311 (
1312 Arc::new(Field::new("metadata", DataType::Binary, false)),
1313 Arc::new(BinaryArray::from(vec![Some(b"m".as_ref())])) as ArrayRef,
1314 ),
1315 (
1316 Arc::new(Field::new("value", DataType::Binary, false)),
1317 Arc::new(BinaryArray::from(vec![Some(b"v".as_ref())])) as ArrayRef,
1318 ),
1319 ])) as ArrayRef;
1320
1321 let struct_array = Arc::new(StructArray::from(vec![(
1322 Arc::new(
1323 Field::new("v", variant_child.data_type().clone(), false).with_metadata(
1324 HashMap::from([(PARQUET_FIELD_ID_META_KEY.to_string(), "1".to_string())]),
1325 ),
1326 ),
1327 variant_child,
1328 )])) as ArrayRef;
1329
1330 let ty = StructType::new(vec![
1331 NestedField::required(1, "v", Type::Variant(VariantType)).into(),
1332 ]);
1333
1334 let err = arrow_struct_to_literal(&struct_array, &ty).unwrap_err();
1335 assert_eq!(err.kind(), ErrorKind::FeatureUnsupported);
1336 assert!(
1337 err.to_string()
1338 .contains("Converting variant Arrow array to Iceberg literal is not supported yet"),
1339 "{err}"
1340 );
1341 }
1342
1343 #[test]
1344 fn test_find_field_by_id() {
1345 let field_a_array = Int32Array::from(vec![Some(42), Some(43), None]);
1347 let field_b_array = StringArray::from(vec![Some("value1"), Some("value2"), None]);
1348
1349 let nested_struct_array =
1351 Arc::new(StructArray::from(vec![
1352 (
1353 Arc::new(Field::new("field_a", DataType::Int32, true).with_metadata(
1354 HashMap::from([(PARQUET_FIELD_ID_META_KEY.to_string(), "1".to_string())]),
1355 )),
1356 Arc::new(field_a_array) as ArrayRef,
1357 ),
1358 (
1359 Arc::new(Field::new("field_b", DataType::Utf8, true).with_metadata(
1360 HashMap::from([(PARQUET_FIELD_ID_META_KEY.to_string(), "2".to_string())]),
1361 )),
1362 Arc::new(field_b_array) as ArrayRef,
1363 ),
1364 ])) as ArrayRef;
1365
1366 let field_c_array = Int32Array::from(vec![Some(100), Some(200), None]);
1367
1368 let struct_array = Arc::new(StructArray::from(vec![
1370 (
1371 Arc::new(
1372 Field::new(
1373 "nested_struct",
1374 DataType::Struct(Fields::from(vec![
1375 Field::new("field_a", DataType::Int32, true).with_metadata(
1376 HashMap::from([(
1377 PARQUET_FIELD_ID_META_KEY.to_string(),
1378 "1".to_string(),
1379 )]),
1380 ),
1381 Field::new("field_b", DataType::Utf8, true).with_metadata(
1382 HashMap::from([(
1383 PARQUET_FIELD_ID_META_KEY.to_string(),
1384 "2".to_string(),
1385 )]),
1386 ),
1387 ])),
1388 true,
1389 )
1390 .with_metadata(HashMap::from([(
1391 PARQUET_FIELD_ID_META_KEY.to_string(),
1392 "3".to_string(),
1393 )])),
1394 ),
1395 nested_struct_array,
1396 ),
1397 (
1398 Arc::new(Field::new("field_c", DataType::Int32, true).with_metadata(
1399 HashMap::from([(PARQUET_FIELD_ID_META_KEY.to_string(), "4".to_string())]),
1400 )),
1401 Arc::new(field_c_array) as ArrayRef,
1402 ),
1403 ])) as ArrayRef;
1404
1405 let accessor = ArrowArrayAccessor::new_with_match_mode(FieldMatchMode::Id);
1407
1408 let nested_field = NestedField::optional(
1410 3,
1411 "nested_struct",
1412 Type::Struct(StructType::new(vec![
1413 Arc::new(NestedField::optional(
1414 1,
1415 "field_a",
1416 Type::Primitive(PrimitiveType::Int),
1417 )),
1418 Arc::new(NestedField::optional(
1419 2,
1420 "field_b",
1421 Type::Primitive(PrimitiveType::String),
1422 )),
1423 ])),
1424 );
1425 let nested_partner = accessor
1426 .field_partner(&struct_array, &nested_field)
1427 .unwrap();
1428
1429 let field_a = NestedField::optional(1, "field_a", Type::Primitive(PrimitiveType::Int));
1431 let field_a_partner = accessor.field_partner(nested_partner, &field_a).unwrap();
1432
1433 let int_array = field_a_partner
1435 .as_any()
1436 .downcast_ref::<Int32Array>()
1437 .unwrap();
1438 assert_eq!(int_array.value(0), 42);
1439 assert_eq!(int_array.value(1), 43);
1440 assert!(int_array.is_null(2));
1441 }
1442
1443 #[test]
1444 fn test_find_field_by_name() {
1445 let field_a_array = Int32Array::from(vec![Some(42), Some(43), None]);
1447 let field_b_array = StringArray::from(vec![Some("value1"), Some("value2"), None]);
1448
1449 let nested_struct_array = Arc::new(StructArray::from(vec![
1451 (
1452 Arc::new(Field::new("field_a", DataType::Int32, true)),
1453 Arc::new(field_a_array) as ArrayRef,
1454 ),
1455 (
1456 Arc::new(Field::new("field_b", DataType::Utf8, true)),
1457 Arc::new(field_b_array) as ArrayRef,
1458 ),
1459 ])) as ArrayRef;
1460
1461 let field_c_array = Int32Array::from(vec![Some(100), Some(200), None]);
1462
1463 let struct_array = Arc::new(StructArray::from(vec![
1465 (
1466 Arc::new(Field::new(
1467 "nested_struct",
1468 DataType::Struct(Fields::from(vec![
1469 Field::new("field_a", DataType::Int32, true),
1470 Field::new("field_b", DataType::Utf8, true),
1471 ])),
1472 true,
1473 )),
1474 nested_struct_array,
1475 ),
1476 (
1477 Arc::new(Field::new("field_c", DataType::Int32, true)),
1478 Arc::new(field_c_array) as ArrayRef,
1479 ),
1480 ])) as ArrayRef;
1481
1482 let accessor = ArrowArrayAccessor::new_with_match_mode(FieldMatchMode::Name);
1484
1485 let nested_field = NestedField::optional(
1487 3,
1488 "nested_struct",
1489 Type::Struct(StructType::new(vec![
1490 Arc::new(NestedField::optional(
1491 1,
1492 "field_a",
1493 Type::Primitive(PrimitiveType::Int),
1494 )),
1495 Arc::new(NestedField::optional(
1496 2,
1497 "field_b",
1498 Type::Primitive(PrimitiveType::String),
1499 )),
1500 ])),
1501 );
1502 let nested_partner = accessor
1503 .field_partner(&struct_array, &nested_field)
1504 .unwrap();
1505
1506 let field_a = NestedField::optional(1, "field_a", Type::Primitive(PrimitiveType::Int));
1508 let field_a_partner = accessor.field_partner(nested_partner, &field_a).unwrap();
1509
1510 let int_array = field_a_partner
1512 .as_any()
1513 .downcast_ref::<Int32Array>()
1514 .unwrap();
1515 assert_eq!(int_array.value(0), 42);
1516 assert_eq!(int_array.value(1), 43);
1517 assert!(int_array.is_null(2));
1518 }
1519
1520 #[test]
1521 fn test_complex_nested() {
1522 let struct_type = StructType::new(vec![
1529 Arc::new(NestedField::required(
1530 0,
1531 "A",
1532 Type::List(ListType::new(Arc::new(NestedField::required(
1533 1,
1534 "item",
1535 Type::Struct(StructType::new(vec![
1536 Arc::new(NestedField::required(
1537 2,
1538 "a1",
1539 Type::Primitive(PrimitiveType::Int),
1540 )),
1541 Arc::new(NestedField::required(
1542 3,
1543 "a2",
1544 Type::Primitive(PrimitiveType::Int),
1545 )),
1546 ])),
1547 )))),
1548 )),
1549 Arc::new(NestedField::required(
1550 4,
1551 "B",
1552 Type::List(ListType::new(Arc::new(NestedField::required(
1553 5,
1554 "item",
1555 Type::Map(MapType::new(
1556 NestedField::optional(6, "keys", Type::Primitive(PrimitiveType::Int))
1557 .into(),
1558 NestedField::optional(7, "values", Type::Primitive(PrimitiveType::Int))
1559 .into(),
1560 )),
1561 )))),
1562 )),
1563 Arc::new(NestedField::required(
1564 8,
1565 "C",
1566 Type::List(ListType::new(Arc::new(NestedField::required(
1567 9,
1568 "item",
1569 Type::List(ListType::new(Arc::new(NestedField::optional(
1570 10,
1571 "item",
1572 Type::Primitive(PrimitiveType::Int),
1573 )))),
1574 )))),
1575 )),
1576 ]);
1577
1578 let struct_array =
1584 {
1585 let a_struct_a1_builder = Int32Builder::new();
1586 let a_struct_a2_builder = Int32Builder::new();
1587 let a_struct_builder =
1588 StructBuilder::new(
1589 vec![
1590 Field::new("a1", DataType::Int32, false).with_metadata(HashMap::from(
1591 [(PARQUET_FIELD_ID_META_KEY.to_string(), "2".to_string())],
1592 )),
1593 Field::new("a2", DataType::Int32, false).with_metadata(HashMap::from(
1594 [(PARQUET_FIELD_ID_META_KEY.to_string(), "3".to_string())],
1595 )),
1596 ],
1597 vec![Box::new(a_struct_a1_builder), Box::new(a_struct_a2_builder)],
1598 );
1599 let a_builder = ListBuilder::new(a_struct_builder);
1600
1601 let map_key_builder = Int32Builder::new();
1602 let map_value_builder = Int32Builder::new();
1603 let map_builder = MapBuilder::new(None, map_key_builder, map_value_builder);
1604 let b_builder = ListBuilder::new(map_builder);
1605
1606 let inner_list_item_builder = Int32Builder::new();
1607 let inner_list_builder = ListBuilder::new(inner_list_item_builder);
1608 let c_builder = ListBuilder::new(inner_list_builder);
1609
1610 let mut top_struct_builder = {
1611 let a_struct_type =
1612 DataType::Struct(Fields::from(vec![
1613 Field::new("a1", DataType::Int32, false).with_metadata(HashMap::from(
1614 [(PARQUET_FIELD_ID_META_KEY.to_string(), "2".to_string())],
1615 )),
1616 Field::new("a2", DataType::Int32, false).with_metadata(HashMap::from(
1617 [(PARQUET_FIELD_ID_META_KEY.to_string(), "3".to_string())],
1618 )),
1619 ]));
1620 let a_type =
1621 DataType::List(Arc::new(Field::new("item", a_struct_type.clone(), true)));
1622
1623 let b_map_entry_struct = Field::new(
1624 "entries",
1625 DataType::Struct(Fields::from(vec![
1626 Field::new("keys", DataType::Int32, false),
1627 Field::new("values", DataType::Int32, true),
1628 ])),
1629 false,
1630 );
1631 let b_map_type =
1632 DataType::Map(Arc::new(b_map_entry_struct), false);
1633 let b_type =
1634 DataType::List(Arc::new(Field::new("item", b_map_type.clone(), true)));
1635
1636 let c_inner_list_type =
1637 DataType::List(Arc::new(Field::new("item", DataType::Int32, true)));
1638 let c_type = DataType::List(Arc::new(Field::new(
1639 "item",
1640 c_inner_list_type.clone(),
1641 true,
1642 )));
1643 StructBuilder::new(
1644 Fields::from(vec![
1645 Field::new("A", a_type.clone(), false).with_metadata(HashMap::from([
1646 (PARQUET_FIELD_ID_META_KEY.to_string(), "0".to_string()),
1647 ])),
1648 Field::new("B", b_type.clone(), false).with_metadata(HashMap::from([
1649 (PARQUET_FIELD_ID_META_KEY.to_string(), "4".to_string()),
1650 ])),
1651 Field::new("C", c_type.clone(), false).with_metadata(HashMap::from([
1652 (PARQUET_FIELD_ID_META_KEY.to_string(), "8".to_string()),
1653 ])),
1654 ]),
1655 vec![
1656 Box::new(a_builder),
1657 Box::new(b_builder),
1658 Box::new(c_builder),
1659 ],
1660 )
1661 };
1662
1663 {
1666 let a_builder = top_struct_builder
1667 .field_builder::<ListBuilder<StructBuilder>>(0)
1668 .unwrap();
1669 let struct_builder = a_builder.values();
1670 struct_builder
1671 .field_builder::<Int32Builder>(0)
1672 .unwrap()
1673 .append_value(10);
1674 struct_builder
1675 .field_builder::<Int32Builder>(1)
1676 .unwrap()
1677 .append_value(20);
1678 struct_builder.append(true);
1679 let struct_builder = a_builder.values();
1680 struct_builder
1681 .field_builder::<Int32Builder>(0)
1682 .unwrap()
1683 .append_value(11);
1684 struct_builder
1685 .field_builder::<Int32Builder>(1)
1686 .unwrap()
1687 .append_value(21);
1688 struct_builder.append(true);
1689 a_builder.append(true);
1690 }
1691 {
1692 let b_builder = top_struct_builder
1693 .field_builder::<ListBuilder<MapBuilder<Int32Builder, Int32Builder>>>(1)
1694 .unwrap();
1695 let map_builder = b_builder.values();
1696 map_builder.keys().append_value(1);
1697 map_builder.values().append_value(100);
1698 map_builder.keys().append_value(3);
1699 map_builder.values().append_value(300);
1700 map_builder.append(true).unwrap();
1701
1702 map_builder.keys().append_value(2);
1703 map_builder.values().append_value(200);
1704 map_builder.append(true).unwrap();
1705
1706 b_builder.append(true);
1707 }
1708 {
1709 let c_builder = top_struct_builder
1710 .field_builder::<ListBuilder<ListBuilder<Int32Builder>>>(2)
1711 .unwrap();
1712 let inner_list_builder = c_builder.values();
1713 inner_list_builder.values().append_value(100);
1714 inner_list_builder.values().append_value(101);
1715 inner_list_builder.values().append_value(102);
1716 inner_list_builder.append(true);
1717 let inner_list_builder = c_builder.values();
1718 inner_list_builder.values().append_value(200);
1719 inner_list_builder.values().append_value(201);
1720 inner_list_builder.append(true);
1721 c_builder.append(true);
1722 }
1723 top_struct_builder.append(true);
1724
1725 {
1728 let a_builder = top_struct_builder
1729 .field_builder::<ListBuilder<StructBuilder>>(0)
1730 .unwrap();
1731 let struct_builder = a_builder.values();
1732 struct_builder
1733 .field_builder::<Int32Builder>(0)
1734 .unwrap()
1735 .append_value(12);
1736 struct_builder
1737 .field_builder::<Int32Builder>(1)
1738 .unwrap()
1739 .append_value(22);
1740 struct_builder.append(true);
1741 let struct_builder = a_builder.values();
1742 struct_builder
1743 .field_builder::<Int32Builder>(0)
1744 .unwrap()
1745 .append_value(13);
1746 struct_builder
1747 .field_builder::<Int32Builder>(1)
1748 .unwrap()
1749 .append_value(23);
1750 struct_builder.append(true);
1751 a_builder.append(true);
1752 }
1753 {
1754 let b_builder = top_struct_builder
1755 .field_builder::<ListBuilder<MapBuilder<Int32Builder, Int32Builder>>>(1)
1756 .unwrap();
1757 let map_builder = b_builder.values();
1758 map_builder.keys().append_value(3);
1759 map_builder.values().append_value(300);
1760 map_builder.append(true).unwrap();
1761
1762 b_builder.append(true);
1763 }
1764 {
1765 let c_builder = top_struct_builder
1766 .field_builder::<ListBuilder<ListBuilder<Int32Builder>>>(2)
1767 .unwrap();
1768 let inner_list_builder = c_builder.values();
1769 inner_list_builder.values().append_value(300);
1770 inner_list_builder.values().append_value(301);
1771 inner_list_builder.values().append_value(302);
1772 inner_list_builder.append(true);
1773 let inner_list_builder = c_builder.values();
1774 inner_list_builder.values().append_value(400);
1775 inner_list_builder.values().append_value(401);
1776 inner_list_builder.append(true);
1777 c_builder.append(true);
1778 }
1779 top_struct_builder.append(true);
1780
1781 Arc::new(top_struct_builder.finish()) as ArrayRef
1782 };
1783
1784 let result = arrow_struct_to_literal(&struct_array, &struct_type).unwrap();
1785 assert_eq!(result, vec![
1786 Some(Literal::Struct(Struct::from_iter(vec![
1787 Some(Literal::List(vec![
1788 Some(Literal::Struct(Struct::from_iter(vec![
1789 Some(Literal::int(10)),
1790 Some(Literal::int(20)),
1791 ]))),
1792 Some(Literal::Struct(Struct::from_iter(vec![
1793 Some(Literal::int(11)),
1794 Some(Literal::int(21)),
1795 ]))),
1796 ])),
1797 Some(Literal::List(vec![
1798 Some(Literal::Map(Map::from_iter(vec![
1799 (Literal::int(1), Some(Literal::int(100))),
1800 (Literal::int(3), Some(Literal::int(300))),
1801 ]))),
1802 Some(Literal::Map(Map::from_iter(vec![(
1803 Literal::int(2),
1804 Some(Literal::int(200))
1805 ),]))),
1806 ])),
1807 Some(Literal::List(vec![
1808 Some(Literal::List(vec![
1809 Some(Literal::int(100)),
1810 Some(Literal::int(101)),
1811 Some(Literal::int(102)),
1812 ])),
1813 Some(Literal::List(vec![
1814 Some(Literal::int(200)),
1815 Some(Literal::int(201)),
1816 ])),
1817 ])),
1818 ]))),
1819 Some(Literal::Struct(Struct::from_iter(vec![
1820 Some(Literal::List(vec![
1821 Some(Literal::Struct(Struct::from_iter(vec![
1822 Some(Literal::int(12)),
1823 Some(Literal::int(22)),
1824 ]))),
1825 Some(Literal::Struct(Struct::from_iter(vec![
1826 Some(Literal::int(13)),
1827 Some(Literal::int(23)),
1828 ]))),
1829 ])),
1830 Some(Literal::List(vec![Some(Literal::Map(Map::from_iter(
1831 vec![(Literal::int(3), Some(Literal::int(300))),]
1832 ))),])),
1833 Some(Literal::List(vec![
1834 Some(Literal::List(vec![
1835 Some(Literal::int(300)),
1836 Some(Literal::int(301)),
1837 Some(Literal::int(302)),
1838 ])),
1839 Some(Literal::List(vec![
1840 Some(Literal::int(400)),
1841 Some(Literal::int(401)),
1842 ])),
1843 ])),
1844 ]))),
1845 ]);
1846 }
1847
1848 #[test]
1849 fn test_create_decimal_array_respects_precision() {
1850 let target_precision = 18u8;
1853 let target_scale = 10i8;
1854 let target_type = DataType::Decimal128(target_precision, target_scale);
1855 let value = PrimitiveLiteral::Int128(10000000000);
1856
1857 let array = create_primitive_array_single_element(&target_type, &Some(value))
1858 .expect("Failed to create decimal array");
1859
1860 match array.data_type() {
1861 DataType::Decimal128(precision, scale) => {
1862 assert_eq!(*precision, target_precision);
1863 assert_eq!(*scale, target_scale);
1864 }
1865 other => panic!("Expected Decimal128, got {other:?}"),
1866 }
1867 }
1868
1869 #[test]
1870 fn test_create_decimal_array_repeated_respects_precision() {
1871 let target_precision = 18u8;
1873 let target_scale = 10i8;
1874 let target_type = DataType::Decimal128(target_precision, target_scale);
1875 let value = PrimitiveLiteral::Int128(10000000000);
1876 let num_rows = 5;
1877
1878 let array = create_primitive_array_repeated(&target_type, &Some(value), num_rows)
1879 .expect("Failed to create repeated decimal array");
1880
1881 match array.data_type() {
1882 DataType::Decimal128(precision, scale) => {
1883 assert_eq!(*precision, target_precision);
1884 assert_eq!(*scale, target_scale);
1885 }
1886 other => panic!("Expected Decimal128, got {other:?}"),
1887 }
1888
1889 assert_eq!(array.len(), num_rows);
1890 }
1891
1892 #[test]
1893 fn test_create_timestamp_microsecond_array_repeated() {
1894 let target_type = DataType::Timestamp(TimeUnit::Microsecond, None);
1895 let value = PrimitiveLiteral::Long(1_740_600_000_000_000);
1896 let num_rows = 3;
1897
1898 let array = create_primitive_array_repeated(&target_type, &Some(value), num_rows)
1899 .expect("Failed to create repeated timestamp microsecond array");
1900
1901 assert_eq!(array.data_type(), &target_type);
1902 assert_eq!(array.len(), num_rows);
1903 }
1904
1905 #[test]
1906 fn test_create_timestamp_microsecond_with_timezone_array_repeated() {
1907 let target_type = DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into()));
1908 let value = PrimitiveLiteral::Long(1_740_600_000_000_000);
1909 let num_rows = 2;
1910
1911 let array = create_primitive_array_repeated(&target_type, &Some(value), num_rows)
1912 .expect("Failed to create repeated timestamp microsecond array with timezone");
1913
1914 assert_eq!(array.data_type(), &target_type);
1915 assert_eq!(array.len(), num_rows);
1916 }
1917}