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