1#![allow(clippy::enum_clike_unportable_variant)]
18
19use crate::{Array, ArrayRef, make_array};
20use arrow_buffer::bit_chunk_iterator::{BitChunkIterator, BitChunks};
21use arrow_buffer::buffer::NullBuffer;
22use arrow_buffer::{BooleanBuffer, Buffer, MutableBuffer, ScalarBuffer};
23use arrow_data::{ArrayData, ArrayDataBuilder};
24use arrow_schema::{ArrowError, DataType, UnionFields, UnionMode};
25use std::any::Any;
28use std::collections::HashSet;
29use std::sync::Arc;
30
31#[derive(Clone)]
123pub struct UnionArray {
124 data_type: DataType,
125 type_ids: ScalarBuffer<i8>,
126 offsets: Option<ScalarBuffer<i32>>,
127 fields: Vec<Option<ArrayRef>>,
128}
129
130impl UnionArray {
131 pub unsafe fn new_unchecked(
150 fields: UnionFields,
151 type_ids: ScalarBuffer<i8>,
152 offsets: Option<ScalarBuffer<i32>>,
153 children: Vec<ArrayRef>,
154 ) -> Self {
155 let mode = if offsets.is_some() {
156 UnionMode::Dense
157 } else {
158 UnionMode::Sparse
159 };
160
161 let len = type_ids.len();
162 let builder = ArrayData::builder(DataType::Union(fields, mode))
163 .add_buffer(type_ids.into_inner())
164 .child_data(children.into_iter().map(Array::into_data).collect())
165 .len(len);
166
167 let data = match offsets {
168 Some(offsets) => unsafe { builder.add_buffer(offsets.into_inner()).build_unchecked() },
169 None => unsafe { builder.build_unchecked() },
170 };
171 Self::from(data)
172 }
173
174 pub fn try_new(
178 fields: UnionFields,
179 type_ids: ScalarBuffer<i8>,
180 offsets: Option<ScalarBuffer<i32>>,
181 children: Vec<ArrayRef>,
182 ) -> Result<Self, ArrowError> {
183 if fields.len() != children.len() {
185 return Err(ArrowError::InvalidArgumentError(
186 "Union fields length must match child arrays length".to_string(),
187 ));
188 }
189
190 if let Some(offsets) = &offsets {
191 if offsets.len() != type_ids.len() {
193 return Err(ArrowError::InvalidArgumentError(
194 "Type Ids and Offsets lengths must match".to_string(),
195 ));
196 }
197 } else {
198 for child in &children {
200 if child.len() != type_ids.len() {
201 return Err(ArrowError::InvalidArgumentError(
202 "Sparse union child arrays must be equal in length to the length of the union".to_string(),
203 ));
204 }
205 }
206 }
207
208 let max_id = fields.iter().map(|(i, _)| i).max().unwrap_or_default() as usize;
210 let mut array_lens = vec![i32::MIN; max_id + 1];
211 for (cd, (field_id, _)) in children.iter().zip(fields.iter()) {
212 array_lens[field_id as usize] = cd.len() as i32;
213 }
214
215 for id in &type_ids {
217 match array_lens.get(*id as usize) {
218 Some(x) if *x != i32::MIN => {}
219 _ => {
220 return Err(ArrowError::InvalidArgumentError(
221 "Type Ids values must match one of the field type ids".to_owned(),
222 ));
223 }
224 }
225 }
226
227 if let Some(offsets) = &offsets {
229 let mut iter = type_ids.iter().zip(offsets.iter());
230 if iter.any(|(type_id, &offset)| offset < 0 || offset >= array_lens[*type_id as usize])
231 {
232 return Err(ArrowError::InvalidArgumentError(
233 "Offsets must be non-negative and within the length of the Array".to_owned(),
234 ));
235 }
236 }
237
238 let union_array = unsafe { Self::new_unchecked(fields, type_ids, offsets, children) };
241 Ok(union_array)
242 }
243
244 pub fn child(&self, type_id: i8) -> &ArrayRef {
251 assert!((type_id as usize) < self.fields.len());
252 let boxed = &self.fields[type_id as usize];
253 boxed.as_ref().expect("invalid type id")
254 }
255
256 pub fn type_id(&self, index: usize) -> i8 {
262 assert!(index < self.type_ids.len());
263 self.type_ids[index]
264 }
265
266 pub fn type_ids(&self) -> &ScalarBuffer<i8> {
268 &self.type_ids
269 }
270
271 pub fn offsets(&self) -> Option<&ScalarBuffer<i32>> {
273 self.offsets.as_ref()
274 }
275
276 pub fn value_offset(&self, index: usize) -> usize {
282 assert!(index < self.len());
283 match &self.offsets {
284 Some(offsets) => offsets[index] as usize,
285 None => self.offset() + index,
286 }
287 }
288
289 pub fn value(&self, i: usize) -> ArrayRef {
297 let type_id = self.type_id(i);
298 let value_offset = self.value_offset(i);
299 let child = self.child(type_id);
300 child.slice(value_offset, 1)
301 }
302
303 pub fn type_names(&self) -> Vec<&str> {
305 match self.data_type() {
306 DataType::Union(fields, _) => fields
307 .iter()
308 .map(|(_, f)| f.name().as_str())
309 .collect::<Vec<&str>>(),
310 _ => unreachable!("Union array's data type is not a union!"),
311 }
312 }
313
314 pub fn fields(&self) -> &UnionFields {
316 match self.data_type() {
317 DataType::Union(fields, _) => fields,
318 _ => unreachable!("Union array's data type is not a union!"),
319 }
320 }
321
322 pub fn is_dense(&self) -> bool {
324 match self.data_type() {
325 DataType::Union(_, mode) => mode == &UnionMode::Dense,
326 _ => unreachable!("Union array's data type is not a union!"),
327 }
328 }
329
330 pub fn slice(&self, offset: usize, length: usize) -> Self {
335 let (offsets, fields) = match self.offsets.as_ref() {
336 Some(offsets) => (Some(offsets.slice(offset, length)), self.fields.clone()),
338 None => {
340 let fields = self
341 .fields
342 .iter()
343 .map(|x| x.as_ref().map(|x| x.slice(offset, length)))
344 .collect();
345 (None, fields)
346 }
347 };
348
349 Self {
350 data_type: self.data_type.clone(),
351 type_ids: self.type_ids.slice(offset, length),
352 offsets,
353 fields,
354 }
355 }
356
357 pub fn into_parts(
385 self,
386 ) -> (
387 UnionFields,
388 ScalarBuffer<i8>,
389 Option<ScalarBuffer<i32>>,
390 Vec<ArrayRef>,
391 ) {
392 let Self {
393 data_type,
394 type_ids,
395 offsets,
396 mut fields,
397 } = self;
398 match data_type {
399 DataType::Union(union_fields, _) => {
400 let children = union_fields
401 .iter()
402 .map(|(type_id, _)| fields[type_id as usize].take().unwrap())
403 .collect();
404 (union_fields, type_ids, offsets, children)
405 }
406 _ => unreachable!(),
407 }
408 }
409
410 fn mask_sparse_skip_without_nulls(&self, nulls: Vec<(i8, NullBuffer)>) -> BooleanBuffer {
412 let fold = |(with_nulls_selected, union_nulls), (is_field, field_nulls)| {
418 (
419 with_nulls_selected | is_field,
420 union_nulls | (is_field & field_nulls),
421 )
422 };
423
424 self.mask_sparse_helper(
425 nulls,
426 |type_ids_chunk_array, nulls_masks_iters| {
427 let (with_nulls_selected, union_nulls) = nulls_masks_iters
428 .iter_mut()
429 .map(|(field_type_id, field_nulls)| {
430 let field_nulls = field_nulls.next().unwrap();
431 let is_field = selection_mask(type_ids_chunk_array, *field_type_id);
432
433 (is_field, field_nulls)
434 })
435 .fold((0, 0), fold);
436
437 let without_nulls_selected = !with_nulls_selected;
439
440 without_nulls_selected | union_nulls
443 },
444 |type_ids_remainder, bit_chunks| {
445 let (with_nulls_selected, union_nulls) = bit_chunks
446 .iter()
447 .map(|(field_type_id, field_bit_chunks)| {
448 let field_nulls = field_bit_chunks.remainder_bits();
449 let is_field = selection_mask(type_ids_remainder, *field_type_id);
450
451 (is_field, field_nulls)
452 })
453 .fold((0, 0), fold);
454
455 let without_nulls_selected = !with_nulls_selected;
456
457 without_nulls_selected | union_nulls
458 },
459 )
460 }
461
462 fn mask_sparse_skip_fully_null(&self, mut nulls: Vec<(i8, NullBuffer)>) -> BooleanBuffer {
464 let DataType::Union(fields, _) = self.data_type() else {
465 unreachable!("Union array's data type is not a union!")
466 };
467
468 let type_ids = fields.iter().map(|(id, _)| id).collect::<HashSet<_>>();
469 let with_nulls = nulls.iter().map(|(id, _)| *id).collect::<HashSet<_>>();
470
471 let without_nulls_ids = type_ids
472 .difference(&with_nulls)
473 .copied()
474 .collect::<Vec<_>>();
475
476 nulls.retain(|(_, nulls)| nulls.null_count() < nulls.len());
477
478 self.mask_sparse_helper(
483 nulls,
484 |type_ids_chunk_array, nulls_masks_iters| {
485 let union_nulls = nulls_masks_iters.iter_mut().fold(
486 0,
487 |union_nulls, (field_type_id, nulls_iter)| {
488 let field_nulls = nulls_iter.next().unwrap();
489
490 if field_nulls == 0 {
491 union_nulls
492 } else {
493 let is_field = selection_mask(type_ids_chunk_array, *field_type_id);
494
495 union_nulls | (is_field & field_nulls)
496 }
497 },
498 );
499
500 let without_nulls_selected =
502 without_nulls_selected(type_ids_chunk_array, &without_nulls_ids);
503
504 union_nulls | without_nulls_selected
507 },
508 |type_ids_remainder, bit_chunks| {
509 let union_nulls =
510 bit_chunks
511 .iter()
512 .fold(0, |union_nulls, (field_type_id, field_bit_chunks)| {
513 let is_field = selection_mask(type_ids_remainder, *field_type_id);
514 let field_nulls = field_bit_chunks.remainder_bits();
515
516 union_nulls | is_field & field_nulls
517 });
518
519 union_nulls | without_nulls_selected(type_ids_remainder, &without_nulls_ids)
520 },
521 )
522 }
523
524 fn mask_sparse_all_with_nulls_skip_one(&self, nulls: Vec<(i8, NullBuffer)>) -> BooleanBuffer {
526 self.mask_sparse_helper(
533 nulls,
534 |type_ids_chunk_array, nulls_masks_iters| {
535 let (is_not_first, union_nulls) = nulls_masks_iters[1..] .iter_mut()
537 .fold(
538 (0, 0),
539 |(is_not_first, union_nulls), (field_type_id, nulls_iter)| {
540 let field_nulls = nulls_iter.next().unwrap();
541 let is_field = selection_mask(type_ids_chunk_array, *field_type_id);
542
543 (
544 is_not_first | is_field,
545 union_nulls | (is_field & field_nulls),
546 )
547 },
548 );
549
550 let is_first = !is_not_first;
551 let first_nulls = nulls_masks_iters[0].1.next().unwrap();
552
553 (is_first & first_nulls) | union_nulls
554 },
555 |type_ids_remainder, bit_chunks| {
556 bit_chunks
557 .iter()
558 .fold(0, |union_nulls, (field_type_id, field_bit_chunks)| {
559 let field_nulls = field_bit_chunks.remainder_bits();
560 let is_field = selection_mask(type_ids_remainder, *field_type_id);
563
564 union_nulls | (is_field & field_nulls)
565 })
566 },
567 )
568 }
569
570 fn mask_sparse_helper(
573 &self,
574 nulls: Vec<(i8, NullBuffer)>,
575 mut mask_chunk: impl FnMut(&[i8; 64], &mut [(i8, BitChunkIterator)]) -> u64,
576 mask_remainder: impl FnOnce(&[i8], &[(i8, BitChunks)]) -> u64,
577 ) -> BooleanBuffer {
578 let bit_chunks = nulls
579 .iter()
580 .map(|(type_id, nulls)| (*type_id, nulls.inner().bit_chunks()))
581 .collect::<Vec<_>>();
582
583 let mut nulls_masks_iter = bit_chunks
584 .iter()
585 .map(|(type_id, bit_chunks)| (*type_id, bit_chunks.iter()))
586 .collect::<Vec<_>>();
587
588 let chunks_exact = self.type_ids.chunks_exact(64);
589 let remainder = chunks_exact.remainder();
590
591 let chunks = chunks_exact.map(|type_ids_chunk| {
592 let type_ids_chunk_array = <&[i8; 64]>::try_from(type_ids_chunk).unwrap();
593
594 mask_chunk(type_ids_chunk_array, &mut nulls_masks_iter)
595 });
596
597 let mut buffer = unsafe { MutableBuffer::from_trusted_len_iter(chunks) };
600
601 if !remainder.is_empty() {
602 buffer.push(mask_remainder(remainder, &bit_chunks));
603 }
604
605 BooleanBuffer::new(buffer.into(), 0, self.type_ids.len())
606 }
607
608 fn gather_nulls(&self, nulls: Vec<(i8, NullBuffer)>) -> BooleanBuffer {
610 let one_null = NullBuffer::new_null(1);
611 let one_valid = NullBuffer::new_valid(1);
612
613 let mut logical_nulls_array = [(&one_valid, Mask::Zero); 256];
620
621 for (type_id, nulls) in &nulls {
622 if nulls.null_count() == nulls.len() {
623 logical_nulls_array[*type_id as u8 as usize] = (&one_null, Mask::Zero);
625 } else {
626 logical_nulls_array[*type_id as u8 as usize] = (nulls, Mask::Max);
627 }
628 }
629
630 match &self.offsets {
631 Some(offsets) => {
632 assert_eq!(self.type_ids.len(), offsets.len());
633
634 BooleanBuffer::collect_bool(self.type_ids.len(), |i| unsafe {
635 let type_id = *self.type_ids.get_unchecked(i);
637 let offset = *offsets.get_unchecked(i);
639
640 let (nulls, offset_mask) = &logical_nulls_array[type_id as u8 as usize];
641
642 nulls
648 .inner()
649 .value_unchecked(offset as usize & *offset_mask as usize)
650 })
651 }
652 None => {
653 BooleanBuffer::collect_bool(self.type_ids.len(), |index| unsafe {
654 let type_id = *self.type_ids.get_unchecked(index);
656
657 let (nulls, index_mask) = &logical_nulls_array[type_id as u8 as usize];
658
659 nulls.inner().value_unchecked(index & *index_mask as usize)
665 })
666 }
667 }
668 }
669
670 fn fields_logical_nulls(&self) -> Vec<(i8, NullBuffer)> {
673 self.fields
674 .iter()
675 .enumerate()
676 .filter_map(|(type_id, field)| Some((type_id as i8, field.as_ref()?.logical_nulls()?)))
677 .filter(|(_, nulls)| nulls.null_count() > 0)
678 .collect()
679 }
680}
681
682impl From<ArrayData> for UnionArray {
683 fn from(data: ArrayData) -> Self {
684 let (data_type, len, _nulls, offset, buffers, child_data) = data.into_parts();
685
686 let (fields, mode) = match &data_type {
687 DataType::Union(fields, mode) => (fields, mode),
688 d => panic!("UnionArray expected ArrayData with type Union got {d}"),
689 };
690
691 let (type_ids, offsets) = match mode {
692 UnionMode::Sparse => {
693 let [buffer]: [Buffer; 1] = buffers.try_into().expect("1 buffer for type_ids");
694 (ScalarBuffer::new(buffer, offset, len), None)
695 }
696 UnionMode::Dense => {
697 let [type_ids_buffer, offsets_buffer]: [Buffer; 2] = buffers
698 .try_into()
699 .expect("2 buffers for type_ids and offsets");
700 (
701 ScalarBuffer::new(type_ids_buffer, offset, len),
702 Some(ScalarBuffer::new(offsets_buffer, offset, len)),
703 )
704 }
705 };
706
707 let max_id = fields.iter().map(|(i, _)| i).max().unwrap_or_default() as usize;
708 let mut boxed_fields = vec![None; max_id + 1];
709 for (cd, (field_id, _)) in child_data.into_iter().zip(fields.iter()) {
710 boxed_fields[field_id as usize] = Some(make_array(cd));
711 }
712 Self {
713 data_type,
714 type_ids,
715 offsets,
716 fields: boxed_fields,
717 }
718 }
719}
720
721impl From<UnionArray> for ArrayData {
722 fn from(array: UnionArray) -> Self {
723 let len = array.len();
724 let DataType::Union(f, _) = &array.data_type else {
725 unreachable!()
726 };
727 let buffers = match array.offsets {
728 Some(o) => vec![array.type_ids.into_inner(), o.into_inner()],
729 None => vec![array.type_ids.into_inner()],
730 };
731
732 let child = f
733 .iter()
734 .map(|(i, _)| array.fields[i as usize].as_ref().unwrap().to_data())
735 .collect();
736
737 let builder = ArrayDataBuilder::new(array.data_type)
738 .len(len)
739 .buffers(buffers)
740 .child_data(child);
741 unsafe { builder.build_unchecked() }
742 }
743}
744
745unsafe impl Array for UnionArray {
747 fn as_any(&self) -> &dyn Any {
748 self
749 }
750
751 fn to_data(&self) -> ArrayData {
752 self.clone().into()
753 }
754
755 fn into_data(self) -> ArrayData {
756 self.into()
757 }
758
759 fn data_type(&self) -> &DataType {
760 &self.data_type
761 }
762
763 fn slice(&self, offset: usize, length: usize) -> ArrayRef {
764 Arc::new(self.slice(offset, length))
765 }
766
767 fn len(&self) -> usize {
768 self.type_ids.len()
769 }
770
771 fn is_empty(&self) -> bool {
772 self.type_ids.is_empty()
773 }
774
775 fn shrink_to_fit(&mut self) {
776 self.type_ids.shrink_to_fit();
777 if let Some(offsets) = &mut self.offsets {
778 offsets.shrink_to_fit();
779 }
780 for array in self.fields.iter_mut().flatten() {
781 array.shrink_to_fit();
782 }
783 self.fields.shrink_to_fit();
784 }
785
786 fn offset(&self) -> usize {
787 0
788 }
789
790 fn nulls(&self) -> Option<&NullBuffer> {
791 None
792 }
793
794 fn logical_nulls(&self) -> Option<NullBuffer> {
795 let DataType::Union(fields, _) = self.data_type() else {
796 unreachable!()
797 };
798
799 if fields.len() <= 1 {
800 return self.fields.iter().find_map(|field_opt| {
801 field_opt
802 .as_ref()
803 .and_then(|field| field.logical_nulls())
804 .map(|logical_nulls| {
805 if self.is_dense() {
806 self.gather_nulls(vec![(0, logical_nulls)]).into()
807 } else {
808 logical_nulls
809 }
810 })
811 });
812 }
813
814 let logical_nulls = self.fields_logical_nulls();
815
816 if logical_nulls.is_empty() {
817 return None;
818 }
819
820 let fully_null_count = logical_nulls
821 .iter()
822 .filter(|(_, nulls)| nulls.null_count() == nulls.len())
823 .count();
824
825 if fully_null_count == fields.len() {
826 if let Some((_, exactly_sized)) = logical_nulls
827 .iter()
828 .find(|(_, nulls)| nulls.len() == self.len())
829 {
830 return Some(exactly_sized.clone());
831 }
832
833 if let Some((_, bigger)) = logical_nulls
834 .iter()
835 .find(|(_, nulls)| nulls.len() > self.len())
836 {
837 return Some(bigger.slice(0, self.len()));
838 }
839
840 return Some(NullBuffer::new_null(self.len()));
841 }
842
843 let boolean_buffer = match &self.offsets {
844 Some(_) => self.gather_nulls(logical_nulls),
845 None => {
846 let gather_relative_cost = if cfg!(target_feature = "avx2") {
854 10
855 } else if cfg!(target_feature = "sse4.1") {
856 3
857 } else if cfg!(target_arch = "x86") || cfg!(target_arch = "x86_64") {
858 2
860 } else {
861 0
865 };
866
867 let strategies = [
868 (SparseStrategy::Gather, gather_relative_cost, true),
869 (
870 SparseStrategy::MaskAllFieldsWithNullsSkipOne,
871 fields.len() - 1,
872 fields.len() == logical_nulls.len(),
873 ),
874 (
875 SparseStrategy::MaskSkipWithoutNulls,
876 logical_nulls.len(),
877 true,
878 ),
879 (
880 SparseStrategy::MaskSkipFullyNull,
881 fields.len() - fully_null_count,
882 true,
883 ),
884 ];
885
886 let (strategy, _, _) = strategies
887 .iter()
888 .filter(|(_, _, applicable)| *applicable)
889 .min_by_key(|(_, cost, _)| cost)
890 .unwrap();
891
892 match strategy {
893 SparseStrategy::Gather => self.gather_nulls(logical_nulls),
894 SparseStrategy::MaskAllFieldsWithNullsSkipOne => {
895 self.mask_sparse_all_with_nulls_skip_one(logical_nulls)
896 }
897 SparseStrategy::MaskSkipWithoutNulls => {
898 self.mask_sparse_skip_without_nulls(logical_nulls)
899 }
900 SparseStrategy::MaskSkipFullyNull => {
901 self.mask_sparse_skip_fully_null(logical_nulls)
902 }
903 }
904 }
905 };
906
907 let null_buffer = NullBuffer::from(boolean_buffer);
908
909 if null_buffer.null_count() > 0 {
910 Some(null_buffer)
911 } else {
912 None
913 }
914 }
915
916 fn is_nullable(&self) -> bool {
917 self.fields
918 .iter()
919 .flatten()
920 .any(|field| field.is_nullable())
921 }
922
923 fn get_buffer_memory_size(&self) -> usize {
924 let mut sum = self.type_ids.inner().capacity();
925 if let Some(o) = self.offsets.as_ref() {
926 sum += o.inner().capacity()
927 }
928 self.fields
929 .iter()
930 .filter_map(|x| x.as_ref().map(|x| x.get_buffer_memory_size()))
931 .sum::<usize>()
932 + sum
933 }
934
935 fn get_array_memory_size(&self) -> usize {
936 let mut sum = self.type_ids.inner().capacity();
937 if let Some(o) = self.offsets.as_ref() {
938 sum += o.inner().capacity()
939 }
940 std::mem::size_of::<Self>()
941 + self
942 .fields
943 .iter()
944 .filter_map(|x| x.as_ref().map(|x| x.get_array_memory_size()))
945 .sum::<usize>()
946 + sum
947 }
948
949 #[cfg(feature = "pool")]
950 fn claim(&self, pool: &dyn arrow_buffer::MemoryPool) {
951 self.type_ids.claim(pool);
952 if let Some(offsets) = &self.offsets {
953 offsets.claim(pool);
954 }
955 for field in self.fields.iter().flatten() {
956 field.claim(pool);
957 }
958 }
959}
960
961impl std::fmt::Debug for UnionArray {
962 fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
963 let header = if self.is_dense() {
964 "UnionArray(Dense)\n["
965 } else {
966 "UnionArray(Sparse)\n["
967 };
968 writeln!(f, "{header}")?;
969
970 writeln!(f, "-- type id buffer:")?;
971 writeln!(f, "{:?}", self.type_ids)?;
972
973 if let Some(offsets) = &self.offsets {
974 writeln!(f, "-- offsets buffer:")?;
975 writeln!(f, "{offsets:?}")?;
976 }
977
978 let DataType::Union(fields, _) = self.data_type() else {
979 unreachable!()
980 };
981
982 for (type_id, field) in fields.iter() {
983 let child = self.child(type_id);
984 writeln!(
985 f,
986 "-- child {}: \"{}\" ({:?})",
987 type_id,
988 field.name(),
989 field.data_type()
990 )?;
991 std::fmt::Debug::fmt(child, f)?;
992 writeln!(f)?;
993 }
994 writeln!(f, "]")
995 }
996}
997
998enum SparseStrategy {
1003 Gather,
1005 MaskAllFieldsWithNullsSkipOne,
1007 MaskSkipWithoutNulls,
1009 MaskSkipFullyNull,
1011}
1012
1013#[derive(Copy, Clone)]
1014#[repr(usize)]
1015enum Mask {
1016 Zero = 0,
1017 Max = usize::MAX,
1018}
1019
1020fn selection_mask(type_ids_chunk: &[i8], type_id: i8) -> u64 {
1021 type_ids_chunk
1022 .iter()
1023 .copied()
1024 .enumerate()
1025 .fold(0, |packed, (bit_idx, v)| {
1026 packed | (((v == type_id) as u64) << bit_idx)
1027 })
1028}
1029
1030fn without_nulls_selected(type_ids_chunk: &[i8], without_nulls_ids: &[i8]) -> u64 {
1032 without_nulls_ids
1033 .iter()
1034 .fold(0, |fully_valid_selected, field_type_id| {
1035 fully_valid_selected | selection_mask(type_ids_chunk, *field_type_id)
1036 })
1037}
1038
1039#[cfg(test)]
1040mod tests {
1041 use super::*;
1042 use std::collections::HashSet;
1043
1044 use crate::array::Int8Type;
1045 use crate::builder::UnionBuilder;
1046 use crate::cast::AsArray;
1047 use crate::types::{Float32Type, Float64Type, Int32Type, Int64Type};
1048 use crate::{Float64Array, Int32Array, Int64Array, StringArray};
1049 use crate::{Int8Array, RecordBatch};
1050 use arrow_buffer::Buffer;
1051 use arrow_schema::{Field, Schema};
1052
1053 #[test]
1054 fn test_dense_i32() {
1055 let mut builder = UnionBuilder::new_dense();
1056 builder.append::<Int32Type>("a", 1).unwrap();
1057 builder.append::<Int32Type>("b", 2).unwrap();
1058 builder.append::<Int32Type>("c", 3).unwrap();
1059 builder.append::<Int32Type>("a", 4).unwrap();
1060 builder.append::<Int32Type>("c", 5).unwrap();
1061 builder.append::<Int32Type>("a", 6).unwrap();
1062 builder.append::<Int32Type>("b", 7).unwrap();
1063 let union = builder.build().unwrap();
1064
1065 let expected_type_ids = vec![0_i8, 1, 2, 0, 2, 0, 1];
1066 let expected_offsets = vec![0_i32, 0, 0, 1, 1, 2, 1];
1067 let expected_array_values = [1_i32, 2, 3, 4, 5, 6, 7];
1068
1069 assert_eq!(*union.type_ids(), expected_type_ids);
1071 for (i, id) in expected_type_ids.iter().enumerate() {
1072 assert_eq!(id, &union.type_id(i));
1073 }
1074
1075 assert_eq!(*union.offsets().unwrap(), expected_offsets);
1077 for (i, id) in expected_offsets.iter().enumerate() {
1078 assert_eq!(union.value_offset(i), *id as usize);
1079 }
1080
1081 assert_eq!(
1083 *union.child(0).as_primitive::<Int32Type>().values(),
1084 [1_i32, 4, 6]
1085 );
1086 assert_eq!(
1087 *union.child(1).as_primitive::<Int32Type>().values(),
1088 [2_i32, 7]
1089 );
1090 assert_eq!(
1091 *union.child(2).as_primitive::<Int32Type>().values(),
1092 [3_i32, 5]
1093 );
1094
1095 assert_eq!(expected_array_values.len(), union.len());
1096 for (i, expected_value) in expected_array_values.iter().enumerate() {
1097 assert!(!union.is_null(i));
1098 let slot = union.value(i);
1099 let slot = slot.as_any().downcast_ref::<Int32Array>().unwrap();
1100 assert_eq!(slot.len(), 1);
1101 let value = slot.value(0);
1102 assert_eq!(expected_value, &value);
1103 }
1104 }
1105
1106 #[test]
1107 fn slice_union_array_single_field() {
1108 let union_array = {
1111 let mut builder = UnionBuilder::new_dense();
1112 builder.append::<Int32Type>("a", 1).unwrap();
1113 builder.append_null::<Int32Type>("a").unwrap();
1114 builder.append::<Int32Type>("a", 3).unwrap();
1115 builder.append_null::<Int32Type>("a").unwrap();
1116 builder.append::<Int32Type>("a", 4).unwrap();
1117 builder.build().unwrap()
1118 };
1119
1120 let union_slice = union_array.slice(1, 3);
1122 let logical_nulls = union_slice.logical_nulls().unwrap();
1123
1124 assert_eq!(logical_nulls.len(), 3);
1125 assert!(logical_nulls.is_null(0));
1126 assert!(logical_nulls.is_valid(1));
1127 assert!(logical_nulls.is_null(2));
1128 }
1129
1130 #[test]
1131 fn test_dense_i32_large() {
1132 let mut builder = UnionBuilder::new_dense();
1133
1134 let expected_type_ids = vec![0_i8; 1024];
1135 let expected_offsets: Vec<_> = (0..1024).collect();
1136 let expected_array_values: Vec<_> = (1..=1024).collect();
1137
1138 expected_array_values
1139 .iter()
1140 .for_each(|v| builder.append::<Int32Type>("a", *v).unwrap());
1141
1142 let union = builder.build().unwrap();
1143
1144 assert_eq!(*union.type_ids(), expected_type_ids);
1146 for (i, id) in expected_type_ids.iter().enumerate() {
1147 assert_eq!(id, &union.type_id(i));
1148 }
1149
1150 assert_eq!(*union.offsets().unwrap(), expected_offsets);
1152 for (i, id) in expected_offsets.iter().enumerate() {
1153 assert_eq!(union.value_offset(i), *id as usize);
1154 }
1155
1156 for (i, expected_value) in expected_array_values.iter().enumerate() {
1157 assert!(!union.is_null(i));
1158 let slot = union.value(i);
1159 let slot = slot.as_primitive::<Int32Type>();
1160 assert_eq!(slot.len(), 1);
1161 let value = slot.value(0);
1162 assert_eq!(expected_value, &value);
1163 }
1164 }
1165
1166 #[test]
1167 fn test_dense_mixed() {
1168 let mut builder = UnionBuilder::new_dense();
1169 builder.append::<Int32Type>("a", 1).unwrap();
1170 builder.append::<Int64Type>("c", 3).unwrap();
1171 builder.append::<Int32Type>("a", 4).unwrap();
1172 builder.append::<Int64Type>("c", 5).unwrap();
1173 builder.append::<Int32Type>("a", 6).unwrap();
1174 let union = builder.build().unwrap();
1175
1176 assert_eq!(5, union.len());
1177 for i in 0..union.len() {
1178 let slot = union.value(i);
1179 assert!(!union.is_null(i));
1180 match i {
1181 0 => {
1182 let slot = slot.as_any().downcast_ref::<Int32Array>().unwrap();
1183 assert_eq!(slot.len(), 1);
1184 let value = slot.value(0);
1185 assert_eq!(1_i32, value);
1186 }
1187 1 => {
1188 let slot = slot.as_any().downcast_ref::<Int64Array>().unwrap();
1189 assert_eq!(slot.len(), 1);
1190 let value = slot.value(0);
1191 assert_eq!(3_i64, value);
1192 }
1193 2 => {
1194 let slot = slot.as_any().downcast_ref::<Int32Array>().unwrap();
1195 assert_eq!(slot.len(), 1);
1196 let value = slot.value(0);
1197 assert_eq!(4_i32, value);
1198 }
1199 3 => {
1200 let slot = slot.as_any().downcast_ref::<Int64Array>().unwrap();
1201 assert_eq!(slot.len(), 1);
1202 let value = slot.value(0);
1203 assert_eq!(5_i64, value);
1204 }
1205 4 => {
1206 let slot = slot.as_any().downcast_ref::<Int32Array>().unwrap();
1207 assert_eq!(slot.len(), 1);
1208 let value = slot.value(0);
1209 assert_eq!(6_i32, value);
1210 }
1211 _ => unreachable!(),
1212 }
1213 }
1214 }
1215
1216 #[test]
1217 fn test_dense_mixed_with_nulls() {
1218 let mut builder = UnionBuilder::new_dense();
1219 builder.append::<Int32Type>("a", 1).unwrap();
1220 builder.append::<Int64Type>("c", 3).unwrap();
1221 builder.append::<Int32Type>("a", 10).unwrap();
1222 builder.append_null::<Int32Type>("a").unwrap();
1223 builder.append::<Int32Type>("a", 6).unwrap();
1224 let union = builder.build().unwrap();
1225
1226 assert_eq!(5, union.len());
1227 for i in 0..union.len() {
1228 let slot = union.value(i);
1229 match i {
1230 0 => {
1231 let slot = slot.as_any().downcast_ref::<Int32Array>().unwrap();
1232 assert!(!slot.is_null(0));
1233 assert_eq!(slot.len(), 1);
1234 let value = slot.value(0);
1235 assert_eq!(1_i32, value);
1236 }
1237 1 => {
1238 let slot = slot.as_any().downcast_ref::<Int64Array>().unwrap();
1239 assert!(!slot.is_null(0));
1240 assert_eq!(slot.len(), 1);
1241 let value = slot.value(0);
1242 assert_eq!(3_i64, value);
1243 }
1244 2 => {
1245 let slot = slot.as_any().downcast_ref::<Int32Array>().unwrap();
1246 assert!(!slot.is_null(0));
1247 assert_eq!(slot.len(), 1);
1248 let value = slot.value(0);
1249 assert_eq!(10_i32, value);
1250 }
1251 3 => assert!(slot.is_null(0)),
1252 4 => {
1253 let slot = slot.as_any().downcast_ref::<Int32Array>().unwrap();
1254 assert!(!slot.is_null(0));
1255 assert_eq!(slot.len(), 1);
1256 let value = slot.value(0);
1257 assert_eq!(6_i32, value);
1258 }
1259 _ => unreachable!(),
1260 }
1261 }
1262 }
1263
1264 #[test]
1265 fn test_dense_mixed_with_nulls_and_offset() {
1266 let mut builder = UnionBuilder::new_dense();
1267 builder.append::<Int32Type>("a", 1).unwrap();
1268 builder.append::<Int64Type>("c", 3).unwrap();
1269 builder.append::<Int32Type>("a", 10).unwrap();
1270 builder.append_null::<Int32Type>("a").unwrap();
1271 builder.append::<Int32Type>("a", 6).unwrap();
1272 let union = builder.build().unwrap();
1273
1274 let slice = union.slice(2, 3);
1275 let new_union = slice.as_any().downcast_ref::<UnionArray>().unwrap();
1276
1277 assert_eq!(3, new_union.len());
1278 for i in 0..new_union.len() {
1279 let slot = new_union.value(i);
1280 match i {
1281 0 => {
1282 let slot = slot.as_any().downcast_ref::<Int32Array>().unwrap();
1283 assert!(!slot.is_null(0));
1284 assert_eq!(slot.len(), 1);
1285 let value = slot.value(0);
1286 assert_eq!(10_i32, value);
1287 }
1288 1 => assert!(slot.is_null(0)),
1289 2 => {
1290 let slot = slot.as_any().downcast_ref::<Int32Array>().unwrap();
1291 assert!(!slot.is_null(0));
1292 assert_eq!(slot.len(), 1);
1293 let value = slot.value(0);
1294 assert_eq!(6_i32, value);
1295 }
1296 _ => unreachable!(),
1297 }
1298 }
1299 }
1300
1301 #[test]
1302 fn test_dense_mixed_with_str() {
1303 let string_array = StringArray::from(vec!["foo", "bar", "baz"]);
1304 let int_array = Int32Array::from(vec![5, 6]);
1305 let float_array = Float64Array::from(vec![10.0]);
1306
1307 let type_ids = [1, 0, 0, 2, 0, 1].into_iter().collect::<ScalarBuffer<i8>>();
1308 let offsets = [0, 0, 1, 0, 2, 1]
1309 .into_iter()
1310 .collect::<ScalarBuffer<i32>>();
1311
1312 let fields = [
1313 (0, Arc::new(Field::new("A", DataType::Utf8, false))),
1314 (1, Arc::new(Field::new("B", DataType::Int32, false))),
1315 (2, Arc::new(Field::new("C", DataType::Float64, false))),
1316 ]
1317 .into_iter()
1318 .collect::<UnionFields>();
1319 let children = [
1320 Arc::new(string_array) as Arc<dyn Array>,
1321 Arc::new(int_array),
1322 Arc::new(float_array),
1323 ]
1324 .into_iter()
1325 .collect();
1326 let array =
1327 UnionArray::try_new(fields, type_ids.clone(), Some(offsets.clone()), children).unwrap();
1328
1329 assert_eq!(*array.type_ids(), type_ids);
1331 for (i, id) in type_ids.iter().enumerate() {
1332 assert_eq!(id, &array.type_id(i));
1333 }
1334
1335 assert_eq!(*array.offsets().unwrap(), offsets);
1337 for (i, id) in offsets.iter().enumerate() {
1338 assert_eq!(*id as usize, array.value_offset(i));
1339 }
1340
1341 assert_eq!(6, array.len());
1343
1344 let slot = array.value(0);
1345 let value = slot.as_any().downcast_ref::<Int32Array>().unwrap().value(0);
1346 assert_eq!(5, value);
1347
1348 let slot = array.value(1);
1349 let value = slot
1350 .as_any()
1351 .downcast_ref::<StringArray>()
1352 .unwrap()
1353 .value(0);
1354 assert_eq!("foo", value);
1355
1356 let slot = array.value(2);
1357 let value = slot
1358 .as_any()
1359 .downcast_ref::<StringArray>()
1360 .unwrap()
1361 .value(0);
1362 assert_eq!("bar", value);
1363
1364 let slot = array.value(3);
1365 let value = slot
1366 .as_any()
1367 .downcast_ref::<Float64Array>()
1368 .unwrap()
1369 .value(0);
1370 assert_eq!(10.0, value);
1371
1372 let slot = array.value(4);
1373 let value = slot
1374 .as_any()
1375 .downcast_ref::<StringArray>()
1376 .unwrap()
1377 .value(0);
1378 assert_eq!("baz", value);
1379
1380 let slot = array.value(5);
1381 let value = slot.as_any().downcast_ref::<Int32Array>().unwrap().value(0);
1382 assert_eq!(6, value);
1383 }
1384
1385 #[test]
1386 fn test_sparse_i32() {
1387 let mut builder = UnionBuilder::new_sparse();
1388 builder.append::<Int32Type>("a", 1).unwrap();
1389 builder.append::<Int32Type>("b", 2).unwrap();
1390 builder.append::<Int32Type>("c", 3).unwrap();
1391 builder.append::<Int32Type>("a", 4).unwrap();
1392 builder.append::<Int32Type>("c", 5).unwrap();
1393 builder.append::<Int32Type>("a", 6).unwrap();
1394 builder.append::<Int32Type>("b", 7).unwrap();
1395 let union = builder.build().unwrap();
1396
1397 let expected_type_ids = vec![0_i8, 1, 2, 0, 2, 0, 1];
1398 let expected_array_values = [1_i32, 2, 3, 4, 5, 6, 7];
1399
1400 assert_eq!(*union.type_ids(), expected_type_ids);
1402 for (i, id) in expected_type_ids.iter().enumerate() {
1403 assert_eq!(id, &union.type_id(i));
1404 }
1405
1406 assert!(union.offsets().is_none());
1408
1409 assert_eq!(
1411 *union.child(0).as_primitive::<Int32Type>().values(),
1412 [1_i32, 0, 0, 4, 0, 6, 0],
1413 );
1414 assert_eq!(
1415 *union.child(1).as_primitive::<Int32Type>().values(),
1416 [0_i32, 2_i32, 0, 0, 0, 0, 7]
1417 );
1418 assert_eq!(
1419 *union.child(2).as_primitive::<Int32Type>().values(),
1420 [0_i32, 0, 3_i32, 0, 5, 0, 0]
1421 );
1422
1423 assert_eq!(expected_array_values.len(), union.len());
1424 for (i, expected_value) in expected_array_values.iter().enumerate() {
1425 assert!(!union.is_null(i));
1426 let slot = union.value(i);
1427 let slot = slot.as_any().downcast_ref::<Int32Array>().unwrap();
1428 assert_eq!(slot.len(), 1);
1429 let value = slot.value(0);
1430 assert_eq!(expected_value, &value);
1431 }
1432 }
1433
1434 #[test]
1435 fn test_sparse_mixed() {
1436 let mut builder = UnionBuilder::new_sparse();
1437 builder.append::<Int32Type>("a", 1).unwrap();
1438 builder.append::<Float64Type>("c", 3.0).unwrap();
1439 builder.append::<Int32Type>("a", 4).unwrap();
1440 builder.append::<Float64Type>("c", 5.0).unwrap();
1441 builder.append::<Int32Type>("a", 6).unwrap();
1442 let union = builder.build().unwrap();
1443
1444 let expected_type_ids = vec![0_i8, 1, 0, 1, 0];
1445
1446 assert_eq!(*union.type_ids(), expected_type_ids);
1448 for (i, id) in expected_type_ids.iter().enumerate() {
1449 assert_eq!(id, &union.type_id(i));
1450 }
1451
1452 assert!(union.offsets().is_none());
1454
1455 for i in 0..union.len() {
1456 let slot = union.value(i);
1457 assert!(!union.is_null(i));
1458 match i {
1459 0 => {
1460 let slot = slot.as_any().downcast_ref::<Int32Array>().unwrap();
1461 assert_eq!(slot.len(), 1);
1462 let value = slot.value(0);
1463 assert_eq!(1_i32, value);
1464 }
1465 1 => {
1466 let slot = slot.as_any().downcast_ref::<Float64Array>().unwrap();
1467 assert_eq!(slot.len(), 1);
1468 let value = slot.value(0);
1469 assert_eq!(value, 3_f64);
1470 }
1471 2 => {
1472 let slot = slot.as_any().downcast_ref::<Int32Array>().unwrap();
1473 assert_eq!(slot.len(), 1);
1474 let value = slot.value(0);
1475 assert_eq!(4_i32, value);
1476 }
1477 3 => {
1478 let slot = slot.as_any().downcast_ref::<Float64Array>().unwrap();
1479 assert_eq!(slot.len(), 1);
1480 let value = slot.value(0);
1481 assert_eq!(5_f64, value);
1482 }
1483 4 => {
1484 let slot = slot.as_any().downcast_ref::<Int32Array>().unwrap();
1485 assert_eq!(slot.len(), 1);
1486 let value = slot.value(0);
1487 assert_eq!(6_i32, value);
1488 }
1489 _ => unreachable!(),
1490 }
1491 }
1492 }
1493
1494 #[test]
1495 fn test_sparse_mixed_with_nulls() {
1496 let mut builder = UnionBuilder::new_sparse();
1497 builder.append::<Int32Type>("a", 1).unwrap();
1498 builder.append_null::<Int32Type>("a").unwrap();
1499 builder.append::<Float64Type>("c", 3.0).unwrap();
1500 builder.append::<Int32Type>("a", 4).unwrap();
1501 let union = builder.build().unwrap();
1502
1503 let expected_type_ids = vec![0_i8, 0, 1, 0];
1504
1505 assert_eq!(*union.type_ids(), expected_type_ids);
1507 for (i, id) in expected_type_ids.iter().enumerate() {
1508 assert_eq!(id, &union.type_id(i));
1509 }
1510
1511 assert!(union.offsets().is_none());
1513
1514 for i in 0..union.len() {
1515 let slot = union.value(i);
1516 match i {
1517 0 => {
1518 let slot = slot.as_any().downcast_ref::<Int32Array>().unwrap();
1519 assert!(!slot.is_null(0));
1520 assert_eq!(slot.len(), 1);
1521 let value = slot.value(0);
1522 assert_eq!(1_i32, value);
1523 }
1524 1 => assert!(slot.is_null(0)),
1525 2 => {
1526 let slot = slot.as_any().downcast_ref::<Float64Array>().unwrap();
1527 assert!(!slot.is_null(0));
1528 assert_eq!(slot.len(), 1);
1529 let value = slot.value(0);
1530 assert_eq!(value, 3_f64);
1531 }
1532 3 => {
1533 let slot = slot.as_any().downcast_ref::<Int32Array>().unwrap();
1534 assert!(!slot.is_null(0));
1535 assert_eq!(slot.len(), 1);
1536 let value = slot.value(0);
1537 assert_eq!(4_i32, value);
1538 }
1539 _ => unreachable!(),
1540 }
1541 }
1542 }
1543
1544 #[test]
1545 fn test_sparse_mixed_with_nulls_and_offset() {
1546 let mut builder = UnionBuilder::new_sparse();
1547 builder.append::<Int32Type>("a", 1).unwrap();
1548 builder.append_null::<Int32Type>("a").unwrap();
1549 builder.append::<Float64Type>("c", 3.0).unwrap();
1550 builder.append_null::<Float64Type>("c").unwrap();
1551 builder.append::<Int32Type>("a", 4).unwrap();
1552 let union = builder.build().unwrap();
1553
1554 let slice = union.slice(1, 4);
1555 let new_union = slice.as_any().downcast_ref::<UnionArray>().unwrap();
1556
1557 assert_eq!(4, new_union.len());
1558 for i in 0..new_union.len() {
1559 let slot = new_union.value(i);
1560 match i {
1561 0 => assert!(slot.is_null(0)),
1562 1 => {
1563 let slot = slot.as_primitive::<Float64Type>();
1564 assert!(!slot.is_null(0));
1565 assert_eq!(slot.len(), 1);
1566 let value = slot.value(0);
1567 assert_eq!(value, 3_f64);
1568 }
1569 2 => assert!(slot.is_null(0)),
1570 3 => {
1571 let slot = slot.as_primitive::<Int32Type>();
1572 assert!(!slot.is_null(0));
1573 assert_eq!(slot.len(), 1);
1574 let value = slot.value(0);
1575 assert_eq!(4_i32, value);
1576 }
1577 _ => unreachable!(),
1578 }
1579 }
1580 }
1581
1582 fn test_union_validity(union_array: &UnionArray) {
1583 assert_eq!(union_array.null_count(), 0);
1584
1585 for i in 0..union_array.len() {
1586 assert!(!union_array.is_null(i));
1587 assert!(union_array.is_valid(i));
1588 }
1589 }
1590
1591 #[test]
1592 fn test_union_array_validity() {
1593 let mut builder = UnionBuilder::new_sparse();
1594 builder.append::<Int32Type>("a", 1).unwrap();
1595 builder.append_null::<Int32Type>("a").unwrap();
1596 builder.append::<Float64Type>("c", 3.0).unwrap();
1597 builder.append_null::<Float64Type>("c").unwrap();
1598 builder.append::<Int32Type>("a", 4).unwrap();
1599 let union = builder.build().unwrap();
1600
1601 test_union_validity(&union);
1602
1603 let mut builder = UnionBuilder::new_dense();
1604 builder.append::<Int32Type>("a", 1).unwrap();
1605 builder.append_null::<Int32Type>("a").unwrap();
1606 builder.append::<Float64Type>("c", 3.0).unwrap();
1607 builder.append_null::<Float64Type>("c").unwrap();
1608 builder.append::<Int32Type>("a", 4).unwrap();
1609 let union = builder.build().unwrap();
1610
1611 test_union_validity(&union);
1612 }
1613
1614 #[test]
1615 fn test_type_check() {
1616 let mut builder = UnionBuilder::new_sparse();
1617 builder.append::<Float32Type>("a", 1.0).unwrap();
1618 let err = builder.append::<Int32Type>("a", 1).unwrap_err().to_string();
1619 assert!(
1620 err.contains(
1621 "Attempt to write col \"a\" with type Int32 doesn't match existing type Float32"
1622 ),
1623 "{}",
1624 err
1625 );
1626 }
1627
1628 #[test]
1629 fn slice_union_array() {
1630 fn create_union(mut builder: UnionBuilder) -> UnionArray {
1632 builder.append::<Int32Type>("a", 1).unwrap();
1633 builder.append_null::<Int32Type>("a").unwrap();
1634 builder.append::<Float64Type>("c", 3.0).unwrap();
1635 builder.append_null::<Float64Type>("c").unwrap();
1636 builder.append::<Int32Type>("a", 4).unwrap();
1637 builder.build().unwrap()
1638 }
1639
1640 fn create_batch(union: UnionArray) -> RecordBatch {
1641 let schema = Schema::new(vec![Field::new(
1642 "struct_array",
1643 union.data_type().clone(),
1644 true,
1645 )]);
1646
1647 RecordBatch::try_new(Arc::new(schema), vec![Arc::new(union)]).unwrap()
1648 }
1649
1650 fn test_slice_union(record_batch_slice: RecordBatch) {
1651 let union_slice = record_batch_slice
1652 .column(0)
1653 .as_any()
1654 .downcast_ref::<UnionArray>()
1655 .unwrap();
1656
1657 assert_eq!(union_slice.type_id(0), 0);
1658 assert_eq!(union_slice.type_id(1), 1);
1659 assert_eq!(union_slice.type_id(2), 1);
1660
1661 let slot = union_slice.value(0);
1662 let array = slot.as_primitive::<Int32Type>();
1663 assert_eq!(array.len(), 1);
1664 assert!(array.is_null(0));
1665
1666 let slot = union_slice.value(1);
1667 let array = slot.as_primitive::<Float64Type>();
1668 assert_eq!(array.len(), 1);
1669 assert!(array.is_valid(0));
1670 assert_eq!(array.value(0), 3.0);
1671
1672 let slot = union_slice.value(2);
1673 let array = slot.as_primitive::<Float64Type>();
1674 assert_eq!(array.len(), 1);
1675 assert!(array.is_null(0));
1676 }
1677
1678 let builder = UnionBuilder::new_sparse();
1680 let record_batch = create_batch(create_union(builder));
1681 let record_batch_slice = record_batch.slice(1, 3);
1683 test_slice_union(record_batch_slice);
1684
1685 let builder = UnionBuilder::new_dense();
1687 let record_batch = create_batch(create_union(builder));
1688 let record_batch_slice = record_batch.slice(1, 3);
1690 test_slice_union(record_batch_slice);
1691 }
1692
1693 #[test]
1694 fn test_custom_type_ids() {
1695 let data_type = DataType::Union(
1696 UnionFields::try_new(
1697 vec![8, 4, 9],
1698 vec![
1699 Field::new("strings", DataType::Utf8, false),
1700 Field::new("integers", DataType::Int32, false),
1701 Field::new("floats", DataType::Float64, false),
1702 ],
1703 )
1704 .unwrap(),
1705 UnionMode::Dense,
1706 );
1707
1708 let string_array = StringArray::from(vec!["foo", "bar", "baz"]);
1709 let int_array = Int32Array::from(vec![5, 6, 4]);
1710 let float_array = Float64Array::from(vec![10.0]);
1711
1712 let type_ids = Buffer::from_vec(vec![4_i8, 8, 4, 8, 9, 4, 8]);
1713 let value_offsets = Buffer::from_vec(vec![0_i32, 0, 1, 1, 0, 2, 2]);
1714
1715 let data = ArrayData::builder(data_type)
1716 .len(7)
1717 .buffers(vec![type_ids, value_offsets])
1718 .child_data(vec![
1719 string_array.into_data(),
1720 int_array.into_data(),
1721 float_array.into_data(),
1722 ])
1723 .build()
1724 .unwrap();
1725
1726 let array = UnionArray::from(data);
1727
1728 let v = array.value(0);
1729 assert_eq!(v.data_type(), &DataType::Int32);
1730 assert_eq!(v.len(), 1);
1731 assert_eq!(v.as_primitive::<Int32Type>().value(0), 5);
1732
1733 let v = array.value(1);
1734 assert_eq!(v.data_type(), &DataType::Utf8);
1735 assert_eq!(v.len(), 1);
1736 assert_eq!(v.as_string::<i32>().value(0), "foo");
1737
1738 let v = array.value(2);
1739 assert_eq!(v.data_type(), &DataType::Int32);
1740 assert_eq!(v.len(), 1);
1741 assert_eq!(v.as_primitive::<Int32Type>().value(0), 6);
1742
1743 let v = array.value(3);
1744 assert_eq!(v.data_type(), &DataType::Utf8);
1745 assert_eq!(v.len(), 1);
1746 assert_eq!(v.as_string::<i32>().value(0), "bar");
1747
1748 let v = array.value(4);
1749 assert_eq!(v.data_type(), &DataType::Float64);
1750 assert_eq!(v.len(), 1);
1751 assert_eq!(v.as_primitive::<Float64Type>().value(0), 10.0);
1752
1753 let v = array.value(5);
1754 assert_eq!(v.data_type(), &DataType::Int32);
1755 assert_eq!(v.len(), 1);
1756 assert_eq!(v.as_primitive::<Int32Type>().value(0), 4);
1757
1758 let v = array.value(6);
1759 assert_eq!(v.data_type(), &DataType::Utf8);
1760 assert_eq!(v.len(), 1);
1761 assert_eq!(v.as_string::<i32>().value(0), "baz");
1762 }
1763
1764 #[test]
1765 fn into_parts() {
1766 let mut builder = UnionBuilder::new_dense();
1767 builder.append::<Int32Type>("a", 1).unwrap();
1768 builder.append::<Int8Type>("b", 2).unwrap();
1769 builder.append::<Int32Type>("a", 3).unwrap();
1770 let dense_union = builder.build().unwrap();
1771
1772 let field = [
1773 &Arc::new(Field::new("a", DataType::Int32, false)),
1774 &Arc::new(Field::new("b", DataType::Int8, false)),
1775 ];
1776 let (union_fields, type_ids, offsets, children) = dense_union.into_parts();
1777 assert_eq!(
1778 union_fields
1779 .iter()
1780 .map(|(_, field)| field)
1781 .collect::<Vec<_>>(),
1782 field
1783 );
1784 assert_eq!(type_ids, [0, 1, 0]);
1785 assert!(offsets.is_some());
1786 assert_eq!(offsets.as_ref().unwrap(), &[0, 0, 1]);
1787
1788 let result = UnionArray::try_new(union_fields, type_ids, offsets, children);
1789 assert!(result.is_ok());
1790 assert_eq!(result.unwrap().len(), 3);
1791
1792 let mut builder = UnionBuilder::new_sparse();
1793 builder.append::<Int32Type>("a", 1).unwrap();
1794 builder.append::<Int8Type>("b", 2).unwrap();
1795 builder.append::<Int32Type>("a", 3).unwrap();
1796 let sparse_union = builder.build().unwrap();
1797
1798 let (union_fields, type_ids, offsets, children) = sparse_union.into_parts();
1799 assert_eq!(type_ids, [0, 1, 0]);
1800 assert!(offsets.is_none());
1801
1802 let result = UnionArray::try_new(union_fields, type_ids, offsets, children);
1803 assert!(result.is_ok());
1804 assert_eq!(result.unwrap().len(), 3);
1805 }
1806
1807 #[test]
1808 fn into_parts_custom_type_ids() {
1809 let set_field_type_ids: [i8; 3] = [8, 4, 9];
1810 let data_type = DataType::Union(
1811 UnionFields::try_new(
1812 set_field_type_ids,
1813 [
1814 Field::new("strings", DataType::Utf8, false),
1815 Field::new("integers", DataType::Int32, false),
1816 Field::new("floats", DataType::Float64, false),
1817 ],
1818 )
1819 .unwrap(),
1820 UnionMode::Dense,
1821 );
1822 let string_array = StringArray::from(vec!["foo", "bar", "baz"]);
1823 let int_array = Int32Array::from(vec![5, 6, 4]);
1824 let float_array = Float64Array::from(vec![10.0]);
1825 let type_ids = Buffer::from_vec(vec![4_i8, 8, 4, 8, 9, 4, 8]);
1826 let value_offsets = Buffer::from_vec(vec![0_i32, 0, 1, 1, 0, 2, 2]);
1827 let data = ArrayData::builder(data_type)
1828 .len(7)
1829 .buffers(vec![type_ids, value_offsets])
1830 .child_data(vec![
1831 string_array.into_data(),
1832 int_array.into_data(),
1833 float_array.into_data(),
1834 ])
1835 .build()
1836 .unwrap();
1837 let array = UnionArray::from(data);
1838
1839 let (union_fields, type_ids, offsets, children) = array.into_parts();
1840 assert_eq!(
1841 type_ids.iter().collect::<HashSet<_>>(),
1842 set_field_type_ids.iter().collect::<HashSet<_>>()
1843 );
1844 let result = UnionArray::try_new(union_fields, type_ids, offsets, children);
1845 assert!(result.is_ok());
1846 let array = result.unwrap();
1847 assert_eq!(array.len(), 7);
1848 }
1849
1850 #[test]
1851 fn test_invalid() {
1852 let fields = UnionFields::try_new(
1853 [3, 2],
1854 [
1855 Field::new("a", DataType::Utf8, false),
1856 Field::new("b", DataType::Utf8, false),
1857 ],
1858 )
1859 .unwrap();
1860 let children = vec![
1861 Arc::new(StringArray::from_iter_values(["a", "b"])) as _,
1862 Arc::new(StringArray::from_iter_values(["c", "d"])) as _,
1863 ];
1864
1865 let type_ids = vec![3, 3, 2].into();
1866 let err =
1867 UnionArray::try_new(fields.clone(), type_ids, None, children.clone()).unwrap_err();
1868 assert_eq!(
1869 err.to_string(),
1870 "Invalid argument error: Sparse union child arrays must be equal in length to the length of the union"
1871 );
1872
1873 let type_ids = vec![1, 2].into();
1874 let err =
1875 UnionArray::try_new(fields.clone(), type_ids, None, children.clone()).unwrap_err();
1876 assert_eq!(
1877 err.to_string(),
1878 "Invalid argument error: Type Ids values must match one of the field type ids"
1879 );
1880
1881 let type_ids = vec![7, 2].into();
1882 let err = UnionArray::try_new(fields.clone(), type_ids, None, children).unwrap_err();
1883 assert_eq!(
1884 err.to_string(),
1885 "Invalid argument error: Type Ids values must match one of the field type ids"
1886 );
1887
1888 let children = vec![
1889 Arc::new(StringArray::from_iter_values(["a", "b"])) as _,
1890 Arc::new(StringArray::from_iter_values(["c"])) as _,
1891 ];
1892 let type_ids = ScalarBuffer::from(vec![3_i8, 3, 2]);
1893 let offsets = Some(vec![0, 1, 0].into());
1894 UnionArray::try_new(fields.clone(), type_ids.clone(), offsets, children.clone()).unwrap();
1895
1896 let offsets = Some(vec![0, 1, 1].into());
1897 let err = UnionArray::try_new(fields.clone(), type_ids.clone(), offsets, children.clone())
1898 .unwrap_err();
1899
1900 assert_eq!(
1901 err.to_string(),
1902 "Invalid argument error: Offsets must be non-negative and within the length of the Array"
1903 );
1904
1905 let offsets = Some(vec![0, 1].into());
1906 let err =
1907 UnionArray::try_new(fields.clone(), type_ids.clone(), offsets, children).unwrap_err();
1908
1909 assert_eq!(
1910 err.to_string(),
1911 "Invalid argument error: Type Ids and Offsets lengths must match"
1912 );
1913
1914 let err = UnionArray::try_new(fields.clone(), type_ids, None, vec![]).unwrap_err();
1915
1916 assert_eq!(
1917 err.to_string(),
1918 "Invalid argument error: Union fields length must match child arrays length"
1919 );
1920 }
1921
1922 #[test]
1923 fn test_logical_nulls_fast_paths() {
1924 let array = UnionArray::try_new(UnionFields::empty(), vec![].into(), None, vec![]).unwrap();
1926
1927 assert_eq!(array.logical_nulls(), None);
1928
1929 let fields = UnionFields::try_new(
1930 [1, 3],
1931 [
1932 Field::new("a", DataType::Int8, false), Field::new("b", DataType::Int8, false), ],
1935 )
1936 .unwrap();
1937 let array = UnionArray::try_new(
1938 fields,
1939 vec![1].into(),
1940 None,
1941 vec![
1942 Arc::new(Int8Array::from_value(5, 1)),
1943 Arc::new(Int8Array::from_value(5, 1)),
1944 ],
1945 )
1946 .unwrap();
1947
1948 assert_eq!(array.logical_nulls(), None);
1949
1950 let nullable_fields = UnionFields::try_new(
1951 [1, 3],
1952 [
1953 Field::new("a", DataType::Int8, true), Field::new("b", DataType::Int8, true), ],
1956 )
1957 .unwrap();
1958 let array = UnionArray::try_new(
1959 nullable_fields.clone(),
1960 vec![1, 1].into(),
1961 None,
1962 vec![
1963 Arc::new(Int8Array::from_value(-5, 2)), Arc::new(Int8Array::from_value(-5, 2)), ],
1966 )
1967 .unwrap();
1968
1969 assert_eq!(array.logical_nulls(), None);
1970
1971 let array = UnionArray::try_new(
1972 nullable_fields.clone(),
1973 vec![1, 1].into(),
1974 None,
1975 vec![
1976 Arc::new(Int8Array::new_null(2)), Arc::new(Int8Array::new_null(2)), ],
1980 )
1981 .unwrap();
1982
1983 assert_eq!(array.logical_nulls(), Some(NullBuffer::new_null(2)));
1984
1985 let array = UnionArray::try_new(
1986 nullable_fields.clone(),
1987 vec![1, 1].into(),
1988 Some(vec![0, 1].into()),
1989 vec![
1990 Arc::new(Int8Array::new_null(3)), Arc::new(Int8Array::new_null(3)), ],
1994 )
1995 .unwrap();
1996
1997 assert_eq!(array.logical_nulls(), Some(NullBuffer::new_null(2)));
1998 }
1999
2000 #[test]
2001 fn test_dense_union_logical_nulls_gather() {
2002 let int_array = Int32Array::from(vec![1, 2]);
2004 let float_array = Float64Array::from(vec![Some(3.2), None]);
2005 let str_array = StringArray::new_null(1);
2006 let type_ids = [1, 1, 3, 3, 4, 4].into_iter().collect::<ScalarBuffer<i8>>();
2007 let offsets = [0, 1, 0, 1, 0, 0]
2008 .into_iter()
2009 .collect::<ScalarBuffer<i32>>();
2010
2011 let children = vec![
2012 Arc::new(int_array) as Arc<dyn Array>,
2013 Arc::new(float_array),
2014 Arc::new(str_array),
2015 ];
2016
2017 let array = UnionArray::try_new(union_fields(), type_ids, Some(offsets), children).unwrap();
2018
2019 let expected = BooleanBuffer::from(vec![true, true, true, false, false, false]);
2020
2021 assert_eq!(expected, array.logical_nulls().unwrap().into_inner());
2022 assert_eq!(expected, array.gather_nulls(array.fields_logical_nulls()));
2023 }
2024
2025 #[test]
2026 fn test_sparse_union_logical_nulls_mask_all_nulls_skip_one() {
2027 let fields: UnionFields = [
2028 (1, Arc::new(Field::new("A", DataType::Int32, true))),
2029 (3, Arc::new(Field::new("B", DataType::Float64, true))),
2030 ]
2031 .into_iter()
2032 .collect();
2033
2034 let int_array = Int32Array::new_null(4);
2036 let float_array = Float64Array::from(vec![None, None, Some(3.2), None]);
2037 let type_ids = [1, 1, 3, 3].into_iter().collect::<ScalarBuffer<i8>>();
2038
2039 let children = vec![Arc::new(int_array) as Arc<dyn Array>, Arc::new(float_array)];
2040
2041 let array = UnionArray::try_new(fields.clone(), type_ids, None, children).unwrap();
2042
2043 let expected = BooleanBuffer::from(vec![false, false, true, false]);
2044
2045 assert_eq!(expected, array.logical_nulls().unwrap().into_inner());
2046 assert_eq!(
2047 expected,
2048 array.mask_sparse_all_with_nulls_skip_one(array.fields_logical_nulls())
2049 );
2050
2051 let len = 2 * 64 + 32;
2053
2054 let int_array = Int32Array::new_null(len);
2055 let float_array = Float64Array::from_iter([Some(3.2), None].into_iter().cycle().take(len));
2056 let type_ids = ScalarBuffer::from_iter([1, 1, 3, 3].into_iter().cycle().take(len));
2057
2058 let array = UnionArray::try_new(
2059 fields,
2060 type_ids,
2061 None,
2062 vec![Arc::new(int_array), Arc::new(float_array)],
2063 )
2064 .unwrap();
2065
2066 let expected =
2067 BooleanBuffer::from_iter([false, false, true, false].into_iter().cycle().take(len));
2068
2069 assert_eq!(array.len(), len);
2070 assert_eq!(expected, array.logical_nulls().unwrap().into_inner());
2071 assert_eq!(
2072 expected,
2073 array.mask_sparse_all_with_nulls_skip_one(array.fields_logical_nulls())
2074 );
2075 }
2076
2077 #[test]
2078 fn test_sparse_union_logical_mask_mixed_nulls_skip_fully_valid() {
2079 let int_array = Int32Array::from_value(2, 6);
2081 let float_array = Float64Array::from_value(4.2, 6);
2082 let str_array = StringArray::new_null(6);
2083 let type_ids = [1, 1, 3, 3, 4, 4].into_iter().collect::<ScalarBuffer<i8>>();
2084
2085 let children = vec![
2086 Arc::new(int_array) as Arc<dyn Array>,
2087 Arc::new(float_array),
2088 Arc::new(str_array),
2089 ];
2090
2091 let array = UnionArray::try_new(union_fields(), type_ids, None, children).unwrap();
2092
2093 let expected = BooleanBuffer::from(vec![true, true, true, true, false, false]);
2094
2095 assert_eq!(expected, array.logical_nulls().unwrap().into_inner());
2096 assert_eq!(
2097 expected,
2098 array.mask_sparse_skip_without_nulls(array.fields_logical_nulls())
2099 );
2100
2101 let len = 2 * 64 + 32;
2103
2104 let int_array = Int32Array::from_value(2, len);
2105 let float_array = Float64Array::from_value(4.2, len);
2106 let str_array = StringArray::from_iter([None, Some("a")].into_iter().cycle().take(len));
2107 let type_ids = ScalarBuffer::from_iter([1, 1, 3, 3, 4, 4].into_iter().cycle().take(len));
2108
2109 let children = vec![
2110 Arc::new(int_array) as Arc<dyn Array>,
2111 Arc::new(float_array),
2112 Arc::new(str_array),
2113 ];
2114
2115 let array = UnionArray::try_new(union_fields(), type_ids, None, children).unwrap();
2116
2117 let expected = BooleanBuffer::from_iter(
2118 [true, true, true, true, false, true]
2119 .into_iter()
2120 .cycle()
2121 .take(len),
2122 );
2123
2124 assert_eq!(array.len(), len);
2125 assert_eq!(expected, array.logical_nulls().unwrap().into_inner());
2126 assert_eq!(
2127 expected,
2128 array.mask_sparse_skip_without_nulls(array.fields_logical_nulls())
2129 );
2130 }
2131
2132 #[test]
2133 fn test_sparse_union_logical_mask_mixed_nulls_skip_fully_null() {
2134 let int_array = Int32Array::new_null(6);
2136 let float_array = Float64Array::from_value(4.2, 6);
2137 let str_array = StringArray::new_null(6);
2138 let type_ids = [1, 1, 3, 3, 4, 4].into_iter().collect::<ScalarBuffer<i8>>();
2139
2140 let children = vec![
2141 Arc::new(int_array) as Arc<dyn Array>,
2142 Arc::new(float_array),
2143 Arc::new(str_array),
2144 ];
2145
2146 let array = UnionArray::try_new(union_fields(), type_ids, None, children).unwrap();
2147
2148 let expected = BooleanBuffer::from(vec![false, false, true, true, false, false]);
2149
2150 assert_eq!(expected, array.logical_nulls().unwrap().into_inner());
2151 assert_eq!(
2152 expected,
2153 array.mask_sparse_skip_fully_null(array.fields_logical_nulls())
2154 );
2155
2156 let len = 2 * 64 + 32;
2158
2159 let int_array = Int32Array::new_null(len);
2160 let float_array = Float64Array::from_value(4.2, len);
2161 let str_array = StringArray::new_null(len);
2162 let type_ids = ScalarBuffer::from_iter([1, 1, 3, 3, 4, 4].into_iter().cycle().take(len));
2163
2164 let children = vec![
2165 Arc::new(int_array) as Arc<dyn Array>,
2166 Arc::new(float_array),
2167 Arc::new(str_array),
2168 ];
2169
2170 let array = UnionArray::try_new(union_fields(), type_ids, None, children).unwrap();
2171
2172 let expected = BooleanBuffer::from_iter(
2173 [false, false, true, true, false, false]
2174 .into_iter()
2175 .cycle()
2176 .take(len),
2177 );
2178
2179 assert_eq!(array.len(), len);
2180 assert_eq!(expected, array.logical_nulls().unwrap().into_inner());
2181 assert_eq!(
2182 expected,
2183 array.mask_sparse_skip_fully_null(array.fields_logical_nulls())
2184 );
2185 }
2186
2187 #[test]
2188 fn test_sparse_union_logical_nulls_gather() {
2189 let n_fields = 50;
2190
2191 let non_null = Int32Array::from_value(2, 4);
2192 let mixed = Int32Array::from(vec![None, None, Some(1), None]);
2193 let fully_null = Int32Array::new_null(4);
2194
2195 let array = UnionArray::try_new(
2196 (1..)
2197 .step_by(2)
2198 .map(|i| {
2199 (
2200 i,
2201 Arc::new(Field::new(format!("f{i}"), DataType::Int32, true)),
2202 )
2203 })
2204 .take(n_fields)
2205 .collect(),
2206 vec![1, 3, 3, 5].into(),
2207 None,
2208 [
2209 Arc::new(non_null) as ArrayRef,
2210 Arc::new(mixed),
2211 Arc::new(fully_null),
2212 ]
2213 .into_iter()
2214 .cycle()
2215 .take(n_fields)
2216 .collect(),
2217 )
2218 .unwrap();
2219
2220 let expected = BooleanBuffer::from(vec![true, false, true, false]);
2221
2222 assert_eq!(expected, array.logical_nulls().unwrap().into_inner());
2223 assert_eq!(expected, array.gather_nulls(array.fields_logical_nulls()));
2224 }
2225
2226 fn union_fields() -> UnionFields {
2227 [
2228 (1, Arc::new(Field::new("A", DataType::Int32, true))),
2229 (3, Arc::new(Field::new("B", DataType::Float64, true))),
2230 (4, Arc::new(Field::new("C", DataType::Utf8, true))),
2231 ]
2232 .into_iter()
2233 .collect()
2234 }
2235
2236 #[test]
2237 fn test_is_nullable() {
2238 assert!(!create_union_array(false, false).is_nullable());
2239 assert!(create_union_array(true, false).is_nullable());
2240 assert!(create_union_array(false, true).is_nullable());
2241 assert!(create_union_array(true, true).is_nullable());
2242 }
2243
2244 fn create_union_array(int_nullable: bool, float_nullable: bool) -> UnionArray {
2251 let int_array = if int_nullable {
2252 Int32Array::from(vec![Some(1), None, Some(3)])
2253 } else {
2254 Int32Array::from(vec![1, 2, 3])
2255 };
2256 let float_array = if float_nullable {
2257 Float64Array::from(vec![Some(3.2), None, Some(4.2)])
2258 } else {
2259 Float64Array::from(vec![3.2, 4.2, 5.2])
2260 };
2261 let type_ids = [0, 1, 0].into_iter().collect::<ScalarBuffer<i8>>();
2262 let offsets = [0, 0, 0].into_iter().collect::<ScalarBuffer<i32>>();
2263 let union_fields = [
2264 (0, Arc::new(Field::new("A", DataType::Int32, true))),
2265 (1, Arc::new(Field::new("B", DataType::Float64, true))),
2266 ]
2267 .into_iter()
2268 .collect::<UnionFields>();
2269
2270 let children = vec![Arc::new(int_array) as Arc<dyn Array>, Arc::new(float_array)];
2271
2272 UnionArray::try_new(union_fields, type_ids, Some(offsets), children).unwrap()
2273 }
2274}