1use crate::concat::concat;
21use crate::dictionary::{merge_dictionary_values, should_merge_dictionary_values};
22use arrow_array::builder::{BooleanBufferBuilder, PrimitiveBuilder};
23use arrow_array::cast::AsArray;
24use arrow_array::types::*;
25use arrow_array::*;
26use arrow_buffer::bit_mask::set_bits;
27use arrow_buffer::bit_util;
28use arrow_buffer::{ArrowNativeType, BooleanBuffer, MutableBuffer, NullBuffer, OffsetBuffer};
29use arrow_data::ByteView;
30use arrow_data::transform::MutableArrayData;
31use arrow_schema::{ArrowError, DataType, FieldRef, Fields};
32use std::sync::Arc;
33
34macro_rules! primitive_helper {
35 ($t:ty, $values:ident, $indices:ident, $data_type:ident) => {
36 interleave_primitive::<$t>($values, $indices, $data_type)
37 };
38}
39
40macro_rules! dict_helper {
41 ($t:ty, $values:expr, $indices:expr) => {
42 interleave_dictionaries::<$t>($values, $indices)
43 };
44}
45
46pub fn interleave(
74 values: &[&dyn Array],
75 indices: &[(usize, usize)],
76) -> Result<ArrayRef, ArrowError> {
77 if values.is_empty() {
78 return Err(ArrowError::InvalidArgumentError(
79 "interleave requires input of at least one array".to_string(),
80 ));
81 }
82 let data_type = values[0].data_type();
83
84 for array in values.iter().skip(1) {
85 if array.data_type() != data_type {
86 return Err(ArrowError::InvalidArgumentError(format!(
87 "It is not possible to interleave arrays of different data types ({} and {})",
88 data_type,
89 array.data_type()
90 )));
91 }
92 }
93
94 if indices.is_empty() {
95 return Ok(new_empty_array(data_type));
96 }
97
98 downcast_primitive! {
99 data_type => (primitive_helper, values, indices, data_type),
100 DataType::Utf8 => interleave_bytes::<Utf8Type>(values, indices),
101 DataType::LargeUtf8 => interleave_bytes::<LargeUtf8Type>(values, indices),
102 DataType::Binary => interleave_bytes::<BinaryType>(values, indices),
103 DataType::LargeBinary => interleave_bytes::<LargeBinaryType>(values, indices),
104 DataType::BinaryView => interleave_views::<BinaryViewType>(values, indices),
105 DataType::Utf8View => interleave_views::<StringViewType>(values, indices),
106 DataType::Dictionary(k, _) => downcast_integer! {
107 k.as_ref() => (dict_helper, values, indices),
108 _ => unreachable!("illegal dictionary key type {k}")
109 },
110 DataType::Struct(fields) => interleave_struct(fields, values, indices),
111 DataType::List(field) => interleave_list::<i32>(values, indices, field),
112 DataType::LargeList(field) => interleave_list::<i64>(values, indices, field),
113 DataType::RunEndEncoded(r, _) => match r.data_type() {
114 DataType::Int16 => interleave_run_end::<Int16Type>(values, indices),
115 DataType::Int32 => interleave_run_end::<Int32Type>(values, indices),
116 DataType::Int64 => interleave_run_end::<Int64Type>(values, indices),
117 t => unreachable!("illegal run-end type {t}"),
118 },
119 DataType::ListView(field) => interleave_list_view::<i32>(values, indices, field),
120 DataType::LargeListView(field) => interleave_list_view::<i64>(values, indices, field),
121 _ => interleave_fallback(values, indices)
122 }
123}
124
125struct Interleave<'a, T> {
129 arrays: Vec<&'a T>,
131 nulls: Option<NullBuffer>,
133}
134
135impl<'a, T: Array + 'static> Interleave<'a, T> {
136 fn new(values: &[&'a dyn Array], indices: &'a [(usize, usize)]) -> Self {
137 let mut has_nulls = false;
138 let arrays: Vec<&T> = values
139 .iter()
140 .map(|x| {
141 has_nulls = has_nulls || x.null_count() != 0;
142 x.as_any().downcast_ref().unwrap()
143 })
144 .collect();
145
146 let nulls = match has_nulls {
147 true => {
148 let nulls = BooleanBuffer::collect_bool(indices.len(), |i| {
149 let (a, b) = indices[i];
150 arrays[a].is_valid(b)
151 });
152 Some(nulls.into())
153 }
154 false => None,
155 };
156
157 Self { arrays, nulls }
158 }
159}
160
161fn interleave_primitive<T: ArrowPrimitiveType>(
162 values: &[&dyn Array],
163 indices: &[(usize, usize)],
164 data_type: &DataType,
165) -> Result<ArrayRef, ArrowError> {
166 let interleaved = Interleave::<'_, PrimitiveArray<T>>::new(values, indices);
167 let arrays = &interleaved.arrays;
168 let len = indices.len();
169
170 let mut output = Vec::with_capacity(len);
171 let dst: *mut T::Native = output.as_mut_ptr();
172 let mut base = 0;
173
174 let chunks = indices.chunks_exact(8);
177 let remainder = chunks.remainder();
178 for chunk in chunks {
179 let v0 = arrays[chunk[0].0].value(chunk[0].1);
180 let v1 = arrays[chunk[1].0].value(chunk[1].1);
181 let v2 = arrays[chunk[2].0].value(chunk[2].1);
182 let v3 = arrays[chunk[3].0].value(chunk[3].1);
183 let v4 = arrays[chunk[4].0].value(chunk[4].1);
184 let v5 = arrays[chunk[5].0].value(chunk[5].1);
185 let v6 = arrays[chunk[6].0].value(chunk[6].1);
186 let v7 = arrays[chunk[7].0].value(chunk[7].1);
187
188 debug_assert!(base + 7 < len);
190 unsafe {
191 dst.add(base).write(v0);
192 dst.add(base + 1).write(v1);
193 dst.add(base + 2).write(v2);
194 dst.add(base + 3).write(v3);
195 dst.add(base + 4).write(v4);
196 dst.add(base + 5).write(v5);
197 dst.add(base + 6).write(v6);
198 dst.add(base + 7).write(v7);
199 }
200 base += 8;
201 }
202
203 for idx in remainder {
204 debug_assert!(base < len);
206 unsafe { dst.add(base).write(arrays[idx.0].value(idx.1)) };
207 base += 1;
208 }
209
210 debug_assert!(base == len);
212 unsafe { output.set_len(len) };
213
214 let array = PrimitiveArray::<T>::try_new(output.into(), interleaved.nulls)?;
215 Ok(Arc::new(array.with_data_type(data_type.clone())))
216}
217
218fn interleave_bytes<T: ByteArrayType>(
219 values: &[&dyn Array],
220 indices: &[(usize, usize)],
221) -> Result<ArrayRef, ArrowError> {
222 let interleaved = Interleave::<'_, GenericByteArray<T>>::new(values, indices);
223
224 let mut capacity = 0;
225 let mut offsets = Vec::with_capacity(indices.len() + 1);
226 offsets.push(T::Offset::from_usize(0).unwrap());
227 for (a, b) in indices {
228 let o = interleaved.arrays[*a].value_offsets();
229 let element_len = o[*b + 1].as_usize() - o[*b].as_usize();
230 capacity += element_len;
231 offsets.push(
232 T::Offset::from_usize(capacity)
233 .ok_or_else(|| ArrowError::OffsetOverflowError(capacity))?,
234 );
235 }
236
237 let mut values = Vec::with_capacity(capacity);
238 for (a, b) in indices {
239 values.extend_from_slice(interleaved.arrays[*a].value(*b).as_ref());
240 }
241
242 let array = unsafe {
244 let offsets = OffsetBuffer::new_unchecked(offsets.into());
245 GenericByteArray::<T>::new_unchecked(offsets, values.into(), interleaved.nulls)
246 };
247 Ok(Arc::new(array))
248}
249
250fn interleave_dictionaries<K: ArrowDictionaryKeyType>(
251 arrays: &[&dyn Array],
252 indices: &[(usize, usize)],
253) -> Result<ArrayRef, ArrowError> {
254 let dictionaries: Vec<_> = arrays.iter().map(|x| x.as_dictionary::<K>()).collect();
255 let (should_merge, has_overflow) =
256 should_merge_dictionary_values::<K>(&dictionaries, indices.len());
257 if !should_merge {
258 return if has_overflow {
259 interleave_fallback(arrays, indices)
260 } else {
261 interleave_fallback_dictionary::<K>(&dictionaries, indices)
262 };
263 }
264
265 let masks: Vec<_> = dictionaries
266 .iter()
267 .enumerate()
268 .map(|(a_idx, dictionary)| {
269 let mut key_mask = BooleanBufferBuilder::new_from_buffer(
270 MutableBuffer::new_null(dictionary.len()),
271 dictionary.len(),
272 );
273
274 for (_, key_idx) in indices.iter().filter(|(a, _)| *a == a_idx) {
275 key_mask.set_bit(*key_idx, true);
276 }
277 key_mask.finish()
278 })
279 .collect();
280
281 let merged = merge_dictionary_values(&dictionaries, Some(&masks))?;
282
283 let mut keys = PrimitiveBuilder::<K>::with_capacity(indices.len());
285 for (a, b) in indices {
286 let old_keys: &PrimitiveArray<K> = dictionaries[*a].keys();
287 match old_keys.is_valid(*b) {
288 true => {
289 let old_key = old_keys.values()[*b];
290 keys.append_value(merged.key_mappings[*a][old_key.as_usize()])
291 }
292 false => keys.append_null(),
293 }
294 }
295 let array = unsafe { DictionaryArray::new_unchecked(keys.finish(), merged.values) };
296 Ok(Arc::new(array))
297}
298
299fn interleave_views<T: ByteViewType>(
300 values: &[&dyn Array],
301 indices: &[(usize, usize)],
302) -> Result<ArrayRef, ArrowError> {
303 let interleaved = Interleave::<'_, GenericByteViewArray<T>>::new(values, indices);
304 let mut buffers = Vec::new();
305
306 let mut offsets = Vec::with_capacity(interleaved.arrays.len() + 1);
308 offsets.push(0);
309 let mut total_buffers = 0;
310 for a in interleaved.arrays.iter() {
311 total_buffers += a.data_buffers().len();
312 offsets.push(total_buffers);
313 }
314
315 let mut buffer_to_new_index = vec![None; total_buffers];
317
318 let views: Vec<u128> = indices
319 .iter()
320 .map(|(array_idx, value_idx)| {
321 let array = interleaved.arrays[*array_idx];
322 let view = array.views().get(*value_idx).unwrap();
323 let view_len = *view as u32;
324 if view_len <= 12 {
325 return *view;
326 }
327 let view = ByteView::from(*view);
329 let buffer_to_new_idx = offsets[*array_idx] + view.buffer_index as usize;
330 let new_buffer_idx: u32 =
331 *buffer_to_new_index[buffer_to_new_idx].get_or_insert_with(|| {
332 buffers.push(array.data_buffers()[view.buffer_index as usize].clone());
333 (buffers.len() - 1) as u32
334 });
335 view.with_buffer_index(new_buffer_idx).as_u128()
336 })
337 .collect();
338
339 let array = unsafe {
340 GenericByteViewArray::<T>::new_unchecked(views.into(), buffers, interleaved.nulls)
341 };
342 Ok(Arc::new(array))
343}
344
345fn interleave_struct(
346 fields: &Fields,
347 values: &[&dyn Array],
348 indices: &[(usize, usize)],
349) -> Result<ArrayRef, ArrowError> {
350 let interleaved = Interleave::<'_, StructArray>::new(values, indices);
351
352 if fields.is_empty() {
353 let array = StructArray::try_new_with_length(
354 fields.clone(),
355 vec![],
356 interleaved.nulls,
357 indices.len(),
358 )?;
359 return Ok(Arc::new(array));
360 }
361
362 let struct_fields_array: Result<Vec<_>, _> = (0..fields.len())
363 .map(|i| {
364 let field_values: Vec<&dyn Array> = interleaved
365 .arrays
366 .iter()
367 .map(|x| x.column(i).as_ref())
368 .collect();
369 interleave(&field_values, indices)
370 })
371 .collect();
372
373 let struct_array =
374 StructArray::try_new(fields.clone(), struct_fields_array?, interleaved.nulls)?;
375 Ok(Arc::new(struct_array))
376}
377
378fn interleave_list_primitive_child<O: OffsetSizeTrait, T: ArrowPrimitiveType>(
379 interleaved: &Interleave<'_, GenericListArray<O>>,
380 indices: &[(usize, usize)],
381 capacity: usize,
382 data_type: &DataType,
383) -> ArrayRef {
384 let child_arrays: Vec<&PrimitiveArray<T>> = interleaved
385 .arrays
386 .iter()
387 .map(|list| list.values().as_primitive::<T>())
388 .collect();
389
390 let has_child_nulls = child_arrays.iter().any(|a| a.null_count() > 0);
391
392 let mut values: Vec<T::Native> = Vec::with_capacity(capacity);
394 for &(array, row) in indices {
395 let o = interleaved.arrays[array].value_offsets();
396 let start = o[row].as_usize();
397 let end = o[row + 1].as_usize();
398 if end > start {
399 values.extend_from_slice(&child_arrays[array].values()[start..end]);
400 }
401 }
402
403 let nulls = if has_child_nulls {
407 let null_byte_len = bit_util::ceil(capacity, 8);
408 let mut output_null_buf = MutableBuffer::from_len_zeroed(null_byte_len);
409
410 let mut offset_write = 0;
411 let mut output_null_count = 0usize;
412 for &(array, row) in indices {
413 let o = interleaved.arrays[array].value_offsets();
414 let start = o[row].as_usize();
415 let end = o[row + 1].as_usize();
416 let len = end - start;
417 if len > 0 {
418 match child_arrays[array].nulls() {
419 Some(null_buffer) => {
420 output_null_count += set_bits(
421 output_null_buf.as_slice_mut(),
422 null_buffer.validity(),
423 offset_write,
424 null_buffer.offset() + start,
425 len,
426 );
427 }
428 None => {
429 let buf = output_null_buf.as_slice_mut();
431 (offset_write..offset_write + len).for_each(|i| bit_util::set_bit(buf, i));
432 }
433 }
434 }
435 offset_write += len;
436 }
437
438 if output_null_count > 0 {
439 let bool_buf = BooleanBuffer::new(output_null_buf.into(), 0, capacity);
440 Some(unsafe { NullBuffer::new_unchecked(bool_buf, output_null_count) })
442 } else {
443 None
444 }
445 } else {
446 None
447 };
448
449 Arc::new(PrimitiveArray::<T>::new(values.into(), nulls).with_data_type(data_type.clone()))
450}
451
452fn interleave_list<O: OffsetSizeTrait>(
453 values: &[&dyn Array],
454 indices: &[(usize, usize)],
455 field: &FieldRef,
456) -> Result<ArrayRef, ArrowError> {
457 let interleaved = Interleave::<'_, GenericListArray<O>>::new(values, indices);
458
459 let mut capacity = 0usize;
461 let mut offsets = Vec::with_capacity(indices.len() + 1);
462 offsets.push(O::from_usize(0).unwrap());
463 for (array, row) in indices {
464 let o = interleaved.arrays[*array].value_offsets();
465 let element_len = o[*row + 1].as_usize() - o[*row].as_usize();
466 capacity += element_len;
467 offsets.push(
468 O::from_usize(capacity).ok_or_else(|| ArrowError::OffsetOverflowError(capacity))?,
469 );
470 }
471
472 macro_rules! list_primitive_helper {
474 ($t:ty) => {
475 interleave_list_primitive_child::<O, $t>(
476 &interleaved,
477 indices,
478 capacity,
479 field.data_type(),
480 )
481 };
482 }
483
484 let child_values = downcast_primitive! {
485 field.data_type() => (list_primitive_helper),
489 _ => {
490 let mut child_indices = Vec::with_capacity(capacity);
493 for (array, row) in indices {
494 let list = interleaved.arrays[*array];
495 let start = list.value_offsets()[*row].as_usize();
496 let end = list.value_offsets()[*row + 1].as_usize();
497 child_indices.extend((start..end).map(|i| (*array, i)));
498 }
499
500 let child_arrays: Vec<&dyn Array> = interleaved
501 .arrays
502 .iter()
503 .map(|list| list.values().as_ref())
504 .collect();
505 interleave(&child_arrays, &child_indices)?
506 }
507 };
508
509 let offsets = OffsetBuffer::new(offsets.into());
510 let list_array =
511 GenericListArray::<O>::new(field.clone(), offsets, child_values, interleaved.nulls);
512
513 Ok(Arc::new(list_array))
514}
515
516fn interleave_run_end<R: RunEndIndexType>(
518 values: &[&dyn Array],
519 indices: &[(usize, usize)],
520) -> Result<ArrayRef, ArrowError> {
521 if indices.is_empty() {
522 return Ok(new_empty_array(values[0].data_type()));
523 }
524
525 let n = indices.len();
526 R::Native::from_usize(n).ok_or_else(|| {
527 ArrowError::ComputeError(format!(
528 "interleave_run_end: output length {n} does not fit run-end type"
529 ))
530 })?;
531
532 let runs: Vec<&RunArray<R>> = values.iter().map(|a| a.as_run::<R>()).collect();
533 let value_arrays: Vec<&dyn Array> = runs.iter().map(|r| r.values().as_ref()).collect();
534
535 let mut phys_pairs: Vec<(usize, usize)> = vec![(0, 0); n];
538 let mut grouped: Vec<(Vec<R::Native>, Vec<usize>)> =
539 (0..runs.len()).map(|_| (Vec::new(), Vec::new())).collect();
540 for (out_pos, &(arr, row)) in indices.iter().enumerate() {
541 let row = R::Native::from_usize(row).ok_or_else(|| {
542 ArrowError::InvalidArgumentError(format!(
543 "interleave_run_end: row index {row} not representable as run-end type {}",
544 R::DATA_TYPE
545 ))
546 })?;
547 grouped[arr].0.push(row);
548 grouped[arr].1.push(out_pos);
549 }
550 for (arr_idx, (logical_rows, out_positions)) in grouped.into_iter().enumerate() {
551 let phys = runs[arr_idx].get_physical_indices(&logical_rows)?;
552 for (p, out_pos) in phys.iter().zip(out_positions.iter()) {
553 phys_pairs[*out_pos] = (arr_idx, *p);
554 }
555 }
556
557 let mut run_ends_buf: Vec<R::Native> = Vec::with_capacity(n);
562 let mut dedup_pairs: Vec<(usize, usize)> = Vec::with_capacity(n);
563 dedup_pairs.push(phys_pairs[0]);
564 for i in 1..n {
565 if phys_pairs[i] != phys_pairs[i - 1] {
566 run_ends_buf.push(R::Native::from_usize(i).unwrap());
567 dedup_pairs.push(phys_pairs[i]);
568 }
569 }
570 run_ends_buf.push(R::Native::from_usize(n).unwrap());
571
572 let taken_values = interleave(&value_arrays, &dedup_pairs)?;
573 let run_ends = PrimitiveArray::<R>::from_iter_values(run_ends_buf);
574
575 Ok(Arc::new(RunArray::<R>::try_new(
576 &run_ends,
577 taken_values.as_ref(),
578 )?))
579}
580
581fn interleave_list_view<O: OffsetSizeTrait>(
582 values: &[&dyn Array],
583 indices: &[(usize, usize)],
584 field: &FieldRef,
585) -> Result<ArrayRef, ArrowError> {
586 let interleaved = Interleave::<'_, GenericListViewArray<O>>::new(values, indices);
587
588 let concat_cost: usize = interleaved.arrays.iter().map(|lv| lv.values().len()).sum();
593 let per_row_cost: usize = indices
594 .iter()
595 .map(|&(a, r)| interleaved.arrays[a].sizes()[r].as_usize())
596 .sum();
597
598 if per_row_cost <= concat_cost {
599 interleave_list_view_copy::<O>(&interleaved, indices, field)
600 } else {
601 interleave_list_view_concat::<O>(&interleaved, indices, field)
602 }
603}
604
605fn interleave_list_view_copy<O: OffsetSizeTrait>(
607 interleaved: &Interleave<'_, GenericListViewArray<O>>,
608 indices: &[(usize, usize)],
609 field: &FieldRef,
610) -> Result<ArrayRef, ArrowError> {
611 let mut capacity = 0usize;
612 let mut offsets = Vec::with_capacity(indices.len());
613 let mut sizes = Vec::with_capacity(indices.len());
614 for &(array_idx, row_idx) in indices {
615 let list = interleaved.arrays[array_idx];
616 let size = list.sizes()[row_idx].as_usize();
617 offsets.push(
618 O::from_usize(capacity).ok_or_else(|| ArrowError::OffsetOverflowError(capacity))?,
619 );
620 sizes.push(O::from_usize(size).ok_or_else(|| ArrowError::OffsetOverflowError(size))?);
621 capacity += size;
622 }
623
624 let child_data: Vec<_> = interleaved
625 .arrays
626 .iter()
627 .map(|list| list.values().to_data())
628 .collect();
629 let child_data_refs: Vec<_> = child_data.iter().collect();
630 let mut mutable_child = MutableArrayData::new(child_data_refs, false, capacity);
631 for &(array_idx, row_idx) in indices {
632 let list = interleaved.arrays[array_idx];
633 let start = list.offsets()[row_idx].as_usize();
634 let size = list.sizes()[row_idx].as_usize();
635 if size > 0 {
636 mutable_child.try_extend(array_idx, start, start + size)?;
637 }
638 }
639
640 Ok(Arc::new(GenericListViewArray::<O>::new(
641 field.clone(),
642 offsets.into(),
643 sizes.into(),
644 make_array(mutable_child.freeze()),
645 interleaved.nulls.clone(),
646 )))
647}
648
649fn interleave_list_view_concat<O: OffsetSizeTrait>(
652 interleaved: &Interleave<'_, GenericListViewArray<O>>,
653 indices: &[(usize, usize)],
654 field: &FieldRef,
655) -> Result<ArrayRef, ArrowError> {
656 let child_arrays: Vec<&dyn Array> = interleaved
657 .arrays
658 .iter()
659 .map(|lv| lv.values().as_ref())
660 .collect();
661 let mut base_offsets = Vec::with_capacity(interleaved.arrays.len());
662 let mut running = 0usize;
663 for lv in &interleaved.arrays {
664 base_offsets.push(running);
665 running += lv.values().len();
666 }
667 let combined_values = concat(&child_arrays)?;
668
669 let mut new_offsets = Vec::with_capacity(indices.len());
670 let mut new_sizes = Vec::with_capacity(indices.len());
671 for &(array_idx, row_idx) in indices {
672 let lv = interleaved.arrays[array_idx];
673 let adjusted = lv.offsets()[row_idx].as_usize() + base_offsets[array_idx];
674 new_offsets.push(
675 O::from_usize(adjusted).ok_or_else(|| ArrowError::OffsetOverflowError(adjusted))?,
676 );
677 new_sizes.push(lv.sizes()[row_idx]);
678 }
679
680 Ok(Arc::new(GenericListViewArray::<O>::new(
681 field.clone(),
682 new_offsets.into(),
683 new_sizes.into(),
684 combined_values,
685 interleaved.nulls.clone(),
686 )))
687}
688
689fn interleave_fallback(
691 values: &[&dyn Array],
692 indices: &[(usize, usize)],
693) -> Result<ArrayRef, ArrowError> {
694 let arrays: Vec<_> = values.iter().map(|x| x.to_data()).collect();
695 let arrays: Vec<_> = arrays.iter().collect();
696 let mut array_data = MutableArrayData::new(arrays, false, indices.len());
697
698 let mut cur_array = indices[0].0;
699 let mut start_row_idx = indices[0].1;
700 let mut end_row_idx = start_row_idx + 1;
701
702 for (array, row) in indices.iter().skip(1).copied() {
703 if array == cur_array && row == end_row_idx {
704 end_row_idx += 1;
706 continue;
707 }
708
709 array_data.try_extend(cur_array, start_row_idx, end_row_idx)?;
711
712 cur_array = array;
714 start_row_idx = row;
715 end_row_idx = start_row_idx + 1;
716 }
717
718 array_data.try_extend(cur_array, start_row_idx, end_row_idx)?;
720 Ok(make_array(array_data.freeze()))
721}
722
723fn interleave_fallback_dictionary<K: ArrowDictionaryKeyType>(
733 dictionaries: &[&DictionaryArray<K>],
734 indices: &[(usize, usize)],
735) -> Result<ArrayRef, ArrowError> {
736 let relative_offsets: Vec<usize> = dictionaries
737 .iter()
738 .scan(0usize, |offset, dict| {
739 let current = *offset;
740 *offset += dict.values().len();
741 Some(current)
742 })
743 .collect();
744 let all_values: Vec<&dyn Array> = dictionaries.iter().map(|d| d.values().as_ref()).collect();
745 let concatenated_values = concat(&all_values)?;
746
747 let any_nulls = dictionaries.iter().any(|d| d.keys().nulls().is_some());
748 let (new_keys, nulls) = if any_nulls {
749 let mut has_nulls = false;
750 let new_keys: Vec<K::Native> = indices
751 .iter()
752 .map(|(array, row)| {
753 let old_keys = dictionaries[*array].keys();
754 if old_keys.is_valid(*row) {
755 let old_key = old_keys.values()[*row].as_usize();
756 K::Native::from_usize(relative_offsets[*array] + old_key)
757 .expect("key overflow should be checked by caller")
758 } else {
759 has_nulls = true;
760 K::Native::ZERO
761 }
762 })
763 .collect();
764
765 let nulls = if has_nulls {
766 let null_buffer = BooleanBuffer::collect_bool(indices.len(), |i| {
767 let (array, row) = indices[i];
768 dictionaries[array].keys().is_valid(row)
769 });
770 Some(NullBuffer::new(null_buffer))
771 } else {
772 None
773 };
774 (new_keys, nulls)
775 } else {
776 let new_keys: Vec<K::Native> = indices
777 .iter()
778 .map(|(array, row)| {
779 let old_key = dictionaries[*array].keys().values()[*row].as_usize();
780 K::Native::from_usize(relative_offsets[*array] + old_key)
781 .expect("key overflow should be checked by caller")
782 })
783 .collect();
784 (new_keys, None)
785 };
786
787 let keys_array = PrimitiveArray::<K>::new(new_keys.into(), nulls);
788 let array = unsafe { DictionaryArray::new_unchecked(keys_array, concatenated_values) };
790 Ok(Arc::new(array))
791}
792
793pub fn interleave_record_batch(
838 record_batches: &[&RecordBatch],
839 indices: &[(usize, usize)],
840) -> Result<RecordBatch, ArrowError> {
841 let schema = record_batches[0].schema();
842 let columns = (0..schema.fields().len())
843 .map(|i| {
844 let column_values: Vec<&dyn Array> = record_batches
845 .iter()
846 .map(|batch| batch.column(i).as_ref())
847 .collect();
848 interleave(&column_values, indices)
849 })
850 .collect::<Result<Vec<_>, _>>()?;
851 RecordBatch::try_new(schema, columns)
852}
853
854#[cfg(test)]
855mod tests {
856 use super::*;
857 use arrow_array::Int32RunArray;
858 use arrow_array::builder::{
859 GenericListBuilder, Int32Builder, PrimitiveBuilder, PrimitiveRunBuilder,
860 };
861 use arrow_array::types::{Decimal128Type, Int8Type, TimestampMicrosecondType};
862 use arrow_buffer::ScalarBuffer;
863 use arrow_schema::{Field, TimeUnit};
864
865 #[test]
866 fn test_primitive() {
867 let a = Int32Array::from_iter_values([1, 2, 3, 4]);
868 let b = Int32Array::from_iter_values([5, 6, 7]);
869 let c = Int32Array::from_iter_values([8, 9, 10]);
870 let values = interleave(&[&a, &b, &c], &[(0, 3), (0, 3), (2, 2), (2, 0), (1, 1)]).unwrap();
871 let v = values.as_primitive::<Int32Type>();
872 assert_eq!(v.values(), &[4, 4, 10, 8, 6]);
873 }
874
875 #[test]
876 fn test_primitive_nulls() {
877 let a = Int32Array::from_iter_values([1, 2, 3, 4]);
878 let b = Int32Array::from_iter([Some(1), Some(4), None]);
879 let values = interleave(&[&a, &b], &[(0, 1), (1, 2), (1, 2), (0, 3), (0, 2)]).unwrap();
880 let v: Vec<_> = values.as_primitive::<Int32Type>().into_iter().collect();
881 assert_eq!(&v, &[Some(2), None, None, Some(4), Some(3)])
882 }
883
884 #[test]
885 fn test_primitive_empty() {
886 let a = Int32Array::from_iter_values([1, 2, 3, 4]);
887 let v = interleave(&[&a], &[]).unwrap();
888 assert!(v.is_empty());
889 assert_eq!(v.data_type(), &DataType::Int32);
890 }
891
892 #[test]
893 fn test_strings() {
894 let a = StringArray::from_iter_values(["a", "b", "c"]);
895 let b = StringArray::from_iter_values(["hello", "world", "foo"]);
896 let values = interleave(&[&a, &b], &[(0, 2), (0, 2), (1, 0), (1, 1), (0, 1)]).unwrap();
897 let v = values.as_string::<i32>();
898 let values: Vec<_> = v.into_iter().collect();
899 assert_eq!(
900 &values,
901 &[
902 Some("c"),
903 Some("c"),
904 Some("hello"),
905 Some("world"),
906 Some("b")
907 ]
908 )
909 }
910
911 #[test]
912 fn test_interleave_dictionary() {
913 let a = DictionaryArray::<Int32Type>::from_iter(["a", "b", "c", "a", "b"]);
914 let b = DictionaryArray::<Int32Type>::from_iter(["a", "c", "a", "c", "a"]);
915
916 let values =
918 interleave(&[&a, &b], &[(0, 2), (0, 2), (0, 2), (1, 0), (1, 1), (0, 1)]).unwrap();
919 let v = values.as_dictionary::<Int32Type>();
920 assert_eq!(v.values().len(), 5);
921
922 let vc = v.downcast_dict::<StringArray>().unwrap();
923 let collected: Vec<_> = vc.into_iter().map(Option::unwrap).collect();
924 assert_eq!(&collected, &["c", "c", "c", "a", "c", "b"]);
925
926 let values = interleave(&[&a, &b], &[(0, 2), (0, 2), (1, 1)]).unwrap();
928 let v = values.as_dictionary::<Int32Type>();
929 assert_eq!(v.values().len(), 1);
930
931 let vc = v.downcast_dict::<StringArray>().unwrap();
932 let collected: Vec<_> = vc.into_iter().map(Option::unwrap).collect();
933 assert_eq!(&collected, &["c", "c", "c"]);
934 }
935
936 #[test]
937 fn test_interleave_dictionary_nulls() {
938 let input_1_keys = Int32Array::from_iter_values([0, 2, 1, 3]);
939 let input_1_values = StringArray::from(vec![Some("foo"), None, Some("bar"), Some("fiz")]);
940 let input_1 = DictionaryArray::new(input_1_keys, Arc::new(input_1_values));
941 let input_2: DictionaryArray<Int32Type> = vec![None].into_iter().collect();
942
943 let expected = vec![Some("fiz"), None, None, Some("foo")];
944
945 let values = interleave(
946 &[&input_1 as _, &input_2 as _],
947 &[(0, 3), (0, 2), (1, 0), (0, 0)],
948 )
949 .unwrap();
950 let dictionary = values.as_dictionary::<Int32Type>();
951 let actual: Vec<Option<&str>> = dictionary
952 .downcast_dict::<StringArray>()
953 .unwrap()
954 .into_iter()
955 .collect();
956
957 assert_eq!(actual, expected);
958 }
959
960 #[test]
961 fn test_interleave_dictionary_overflow_same_values() {
962 let values: ArrayRef = Arc::new(StringArray::from_iter_values(
963 (0..50).map(|i| format!("v{i}")),
964 ));
965
966 let dict1 = DictionaryArray::<Int8Type>::new(
972 Int8Array::from_iter_values([0, 1, 2]),
973 values.clone(),
974 );
975 let dict2 = DictionaryArray::<Int8Type>::new(
976 Int8Array::from_iter_values([0, 1, 2]),
977 values.clone(),
978 );
979 let dict3 =
980 DictionaryArray::<Int8Type>::new(Int8Array::from_iter_values([49]), values.clone());
981
982 let indices = &[(0, 0), (1, 0), (2, 0)];
983 let result = interleave(&[&dict1, &dict2, &dict3], indices).unwrap();
984
985 let dict_result = result.as_dictionary::<Int8Type>();
986 let string_result: Vec<_> = dict_result
987 .downcast_dict::<StringArray>()
988 .unwrap()
989 .into_iter()
990 .map(|x| x.unwrap())
991 .collect();
992 assert_eq!(string_result, vec!["v0", "v0", "v49"]);
993 }
994
995 fn test_interleave_lists<O: OffsetSizeTrait>() {
996 let mut a = GenericListBuilder::<O, _>::new(Int32Builder::new());
998 a.values().append_value(1);
999 a.values().append_value(2);
1000 a.append(true);
1001 a.append(false);
1002 a.values().append_value(3);
1003 a.append(true);
1004 let a = a.finish();
1005
1006 let mut b = GenericListBuilder::<O, _>::new(Int32Builder::new());
1008 b.values().append_value(4);
1009 b.append(true);
1010 b.append(false);
1011 b.values().append_value(5);
1012 b.values().append_value(6);
1013 b.values().append_null();
1014 b.append(true);
1015 let b = b.finish();
1016
1017 let values = interleave(&[&a, &b], &[(0, 2), (0, 1), (1, 0), (1, 2), (1, 1)]).unwrap();
1018 let v = values
1019 .as_any()
1020 .downcast_ref::<GenericListArray<O>>()
1021 .unwrap();
1022
1023 let mut expected = GenericListBuilder::<O, _>::new(Int32Builder::new());
1025 expected.values().append_value(3);
1026 expected.append(true);
1027 expected.append(false);
1028 expected.values().append_value(4);
1029 expected.append(true);
1030 expected.values().append_value(5);
1031 expected.values().append_value(6);
1032 expected.values().append_null();
1033 expected.append(true);
1034 expected.append(false);
1035 let expected = expected.finish();
1036
1037 assert_eq!(v, &expected);
1038 }
1039
1040 #[test]
1041 fn test_lists() {
1042 test_interleave_lists::<i32>();
1043 }
1044
1045 #[test]
1046 fn test_large_lists() {
1047 test_interleave_lists::<i64>();
1048 }
1049
1050 type ListRow<T> = Option<Vec<Option<<T as ArrowPrimitiveType>::Native>>>;
1053
1054 fn list_of_primitive<O: OffsetSizeTrait, T: ArrowPrimitiveType>(
1057 data_type: &DataType,
1058 rows: &[ListRow<T>],
1059 ) -> GenericListArray<O> {
1060 let mut builder = GenericListBuilder::<O, _>::new(
1061 PrimitiveBuilder::<T>::new().with_data_type(data_type.clone()),
1062 );
1063 for row in rows {
1064 match row {
1065 Some(items) => {
1066 items
1067 .iter()
1068 .for_each(|v| builder.values().append_option(*v));
1069 builder.append(true);
1070 }
1071 None => builder.append(false),
1072 }
1073 }
1074 builder.finish()
1075 }
1076
1077 fn check_interleave_list_primitive<O: OffsetSizeTrait, T: ArrowPrimitiveType>(
1080 data_type: &DataType,
1081 inputs: &[&[ListRow<T>]],
1082 indices: &[(usize, usize)],
1083 expected: &[ListRow<T>],
1084 ) {
1085 let arrays: Vec<_> = inputs
1086 .iter()
1087 .map(|rows| list_of_primitive::<O, T>(data_type, rows))
1088 .collect();
1089 let refs: Vec<&dyn Array> = arrays.iter().map(|a| a as &dyn Array).collect();
1090
1091 let values = interleave(&refs, indices).unwrap();
1092 let v = values
1093 .as_any()
1094 .downcast_ref::<GenericListArray<O>>()
1095 .unwrap();
1096
1097 assert_eq!(v, &list_of_primitive::<O, T>(data_type, expected));
1098 assert_eq!(v.values().data_type(), data_type);
1101 }
1102
1103 fn test_interleave_lists_decimal<O: OffsetSizeTrait>() {
1104 check_interleave_list_primitive::<O, Decimal128Type>(
1106 &DataType::Decimal128(20, 3),
1107 &[
1108 &[
1109 Some(vec![Some(1), Some(2)]),
1110 None,
1111 Some(vec![Some(3), None]),
1112 ], &[Some(vec![Some(4)]), Some(vec![Some(5), Some(6)])], ],
1115 &[(0, 2), (0, 1), (1, 0), (1, 1)],
1116 &[
1117 Some(vec![Some(3), None]),
1118 None,
1119 Some(vec![Some(4)]),
1120 Some(vec![Some(5), Some(6)]),
1121 ],
1122 );
1123 }
1124
1125 #[test]
1126 fn test_lists_decimal() {
1127 test_interleave_lists_decimal::<i32>();
1128 test_interleave_lists_decimal::<i64>();
1129 }
1130
1131 fn test_interleave_lists_timestamp_tz<O: OffsetSizeTrait>() {
1132 check_interleave_list_primitive::<O, TimestampMicrosecondType>(
1134 &DataType::Timestamp(TimeUnit::Microsecond, Some("+08:00".into())),
1135 &[&[Some(vec![Some(1), Some(2)]), Some(vec![Some(3)])]],
1136 &[(0, 1), (0, 0)],
1137 &[Some(vec![Some(3)]), Some(vec![Some(1), Some(2)])],
1138 );
1139 }
1140
1141 #[test]
1142 fn test_lists_timestamp_tz() {
1143 test_interleave_lists_timestamp_tz::<i32>();
1144 test_interleave_lists_timestamp_tz::<i64>();
1145 }
1146
1147 fn test_interleave_list_views<O: OffsetSizeTrait>() {
1148 let mut a = GenericListBuilder::<O, _>::new(Int32Builder::new());
1150 a.values().append_value(1);
1151 a.values().append_value(2);
1152 a.append(true);
1153 a.append(false);
1154 a.values().append_value(3);
1155 a.append(true);
1156 let a: GenericListViewArray<O> = a.finish().into();
1157
1158 let mut b = GenericListBuilder::<O, _>::new(Int32Builder::new());
1160 b.values().append_value(4);
1161 b.append(true);
1162 b.append(false);
1163 b.values().append_value(5);
1164 b.values().append_value(6);
1165 b.values().append_null();
1166 b.append(true);
1167 let b: GenericListViewArray<O> = b.finish().into();
1168
1169 let values = interleave(&[&a, &b], &[(0, 2), (0, 1), (1, 0), (1, 2), (1, 1)]).unwrap();
1170 let v = values
1171 .as_any()
1172 .downcast_ref::<GenericListViewArray<O>>()
1173 .unwrap();
1174
1175 let mut expected = GenericListBuilder::<O, _>::new(Int32Builder::new());
1177 expected.values().append_value(3);
1178 expected.append(true);
1179 expected.append(false);
1180 expected.values().append_value(4);
1181 expected.append(true);
1182 expected.values().append_value(5);
1183 expected.values().append_value(6);
1184 expected.values().append_null();
1185 expected.append(true);
1186 expected.append(false);
1187 let expected: GenericListViewArray<O> = expected.finish().into();
1188
1189 assert_eq!(v, &expected);
1190 }
1191
1192 #[test]
1193 fn test_list_views() {
1194 test_interleave_list_views::<i32>();
1195 }
1196
1197 #[test]
1198 fn test_large_list_views() {
1199 test_interleave_list_views::<i64>();
1200 }
1201
1202 #[test]
1203 fn test_interleave_list_view_overlapping() {
1204 let field = Arc::new(Field::new_list_field(DataType::Int64, false));
1205
1206 let lv_a = ListViewArray::new(
1210 Arc::clone(&field),
1211 ScalarBuffer::from(vec![0i32, 0, 0, 0, 0, 5, 5, 5, 5, 5]),
1212 ScalarBuffer::from(vec![5i32; 10]),
1213 Arc::new(Int64Array::from_iter_values(0..10)),
1214 None,
1215 );
1216
1217 let lv_b = ListViewArray::new(
1221 Arc::clone(&field),
1222 ScalarBuffer::from(vec![0i32, 0, 0, 0, 3, 3, 3, 3]),
1223 ScalarBuffer::from(vec![3i32; 8]),
1224 Arc::new(Int64Array::from_iter_values(100..106)),
1225 None,
1226 );
1227
1228 let indices: Vec<(usize, usize)> = vec![
1229 (0, 0),
1230 (1, 0),
1231 (0, 5),
1232 (1, 4),
1233 (0, 1),
1234 (1, 1),
1235 (0, 6),
1236 (1, 5),
1237 ];
1238 let result = interleave(&[&lv_a as &dyn Array, &lv_b as &dyn Array], &indices).unwrap();
1239 result
1240 .to_data()
1241 .validate_full()
1242 .expect("result must be valid");
1243
1244 let result_lv = result.as_list_view::<i32>();
1245 assert_eq!(result_lv.len(), 8);
1246 assert_eq!(
1247 result_lv.value(0).as_primitive::<Int64Type>().values(),
1248 &[0, 1, 2, 3, 4]
1249 );
1250 assert_eq!(
1251 result_lv.value(1).as_primitive::<Int64Type>().values(),
1252 &[100, 101, 102]
1253 );
1254 assert_eq!(
1255 result_lv.value(2).as_primitive::<Int64Type>().values(),
1256 &[5, 6, 7, 8, 9]
1257 );
1258 assert_eq!(
1259 result_lv.value(3).as_primitive::<Int64Type>().values(),
1260 &[103, 104, 105]
1261 );
1262
1263 let total_input_elements = lv_a.values().len() + lv_b.values().len();
1266 assert_eq!(result_lv.values().len(), total_input_elements);
1267 }
1268
1269 #[test]
1270 fn test_struct_without_nulls() {
1271 let fields = Fields::from(vec![
1272 Field::new("number_col", DataType::Int32, false),
1273 Field::new("string_col", DataType::Utf8, false),
1274 ]);
1275 let a = {
1276 let number_col = Int32Array::from_iter_values([1, 2, 3, 4]);
1277 let string_col = StringArray::from_iter_values(["a", "b", "c", "d"]);
1278
1279 StructArray::try_new(
1280 fields.clone(),
1281 vec![Arc::new(number_col), Arc::new(string_col)],
1282 None,
1283 )
1284 .unwrap()
1285 };
1286
1287 let b = {
1288 let number_col = Int32Array::from_iter_values([5, 6, 7]);
1289 let string_col = StringArray::from_iter_values(["hello", "world", "foo"]);
1290
1291 StructArray::try_new(
1292 fields.clone(),
1293 vec![Arc::new(number_col), Arc::new(string_col)],
1294 None,
1295 )
1296 .unwrap()
1297 };
1298
1299 let c = {
1300 let number_col = Int32Array::from_iter_values([8, 9, 10]);
1301 let string_col = StringArray::from_iter_values(["x", "y", "z"]);
1302
1303 StructArray::try_new(
1304 fields.clone(),
1305 vec![Arc::new(number_col), Arc::new(string_col)],
1306 None,
1307 )
1308 .unwrap()
1309 };
1310
1311 let values = interleave(&[&a, &b, &c], &[(0, 3), (0, 3), (2, 2), (2, 0), (1, 1)]).unwrap();
1312 let values_struct = values.as_struct();
1313 assert_eq!(values_struct.data_type(), &DataType::Struct(fields));
1314 assert_eq!(values_struct.null_count(), 0);
1315
1316 let values_number = values_struct.column(0).as_primitive::<Int32Type>();
1317 assert_eq!(values_number.values(), &[4, 4, 10, 8, 6]);
1318 let values_string = values_struct.column(1).as_string::<i32>();
1319 let values_string: Vec<_> = values_string.into_iter().collect();
1320 assert_eq!(
1321 &values_string,
1322 &[Some("d"), Some("d"), Some("z"), Some("x"), Some("world")]
1323 );
1324 }
1325
1326 #[test]
1327 fn test_struct_with_nulls_in_values() {
1328 let fields = Fields::from(vec![
1329 Field::new("number_col", DataType::Int32, true),
1330 Field::new("string_col", DataType::Utf8, true),
1331 ]);
1332 let a = {
1333 let number_col = Int32Array::from_iter_values([1, 2, 3, 4]);
1334 let string_col = StringArray::from_iter_values(["a", "b", "c", "d"]);
1335
1336 StructArray::try_new(
1337 fields.clone(),
1338 vec![Arc::new(number_col), Arc::new(string_col)],
1339 None,
1340 )
1341 .unwrap()
1342 };
1343
1344 let b = {
1345 let number_col = Int32Array::from_iter([Some(1), Some(4), None]);
1346 let string_col = StringArray::from(vec![Some("hello"), None, Some("foo")]);
1347
1348 StructArray::try_new(
1349 fields.clone(),
1350 vec![Arc::new(number_col), Arc::new(string_col)],
1351 None,
1352 )
1353 .unwrap()
1354 };
1355
1356 let values = interleave(&[&a, &b], &[(0, 1), (1, 2), (1, 2), (0, 3), (1, 1)]).unwrap();
1357 let values_struct = values.as_struct();
1358 assert_eq!(values_struct.data_type(), &DataType::Struct(fields));
1359
1360 assert_eq!(values_struct.null_count(), 0);
1362
1363 let values_number: Vec<_> = values_struct
1364 .column(0)
1365 .as_primitive::<Int32Type>()
1366 .into_iter()
1367 .collect();
1368 assert_eq!(values_number, &[Some(2), None, None, Some(4), Some(4)]);
1369
1370 let values_string = values_struct.column(1).as_string::<i32>();
1371 let values_string: Vec<_> = values_string.into_iter().collect();
1372 assert_eq!(
1373 &values_string,
1374 &[Some("b"), Some("foo"), Some("foo"), Some("d"), None]
1375 );
1376 }
1377
1378 #[test]
1379 fn test_struct_with_nulls() {
1380 let fields = Fields::from(vec![
1381 Field::new("number_col", DataType::Int32, false),
1382 Field::new("string_col", DataType::Utf8, false),
1383 ]);
1384 let a = {
1385 let number_col = Int32Array::from_iter_values([1, 2, 3, 4]);
1386 let string_col = StringArray::from_iter_values(["a", "b", "c", "d"]);
1387
1388 StructArray::try_new(
1389 fields.clone(),
1390 vec![Arc::new(number_col), Arc::new(string_col)],
1391 None,
1392 )
1393 .unwrap()
1394 };
1395
1396 let b = {
1397 let number_col = Int32Array::from_iter_values([5, 6, 7]);
1398 let string_col = StringArray::from_iter_values(["hello", "world", "foo"]);
1399
1400 StructArray::try_new(
1401 fields.clone(),
1402 vec![Arc::new(number_col), Arc::new(string_col)],
1403 Some(NullBuffer::from(&[true, false, true])),
1404 )
1405 .unwrap()
1406 };
1407
1408 let c = {
1409 let number_col = Int32Array::from_iter_values([8, 9, 10]);
1410 let string_col = StringArray::from_iter_values(["x", "y", "z"]);
1411
1412 StructArray::try_new(
1413 fields.clone(),
1414 vec![Arc::new(number_col), Arc::new(string_col)],
1415 None,
1416 )
1417 .unwrap()
1418 };
1419
1420 let values = interleave(&[&a, &b, &c], &[(0, 3), (0, 3), (2, 2), (1, 1), (2, 0)]).unwrap();
1421 let values_struct = values.as_struct();
1422 assert_eq!(values_struct.data_type(), &DataType::Struct(fields));
1423
1424 let validity: Vec<bool> = {
1425 let null_buffer = values_struct.nulls().expect("should_have_nulls");
1426
1427 null_buffer.iter().collect()
1428 };
1429 assert_eq!(validity, &[true, true, true, false, true]);
1430 let values_number = values_struct.column(0).as_primitive::<Int32Type>();
1431 assert_eq!(values_number.values(), &[4, 4, 10, 6, 8]);
1432 let values_string = values_struct.column(1).as_string::<i32>();
1433 let values_string: Vec<_> = values_string.into_iter().collect();
1434 assert_eq!(
1435 &values_string,
1436 &[Some("d"), Some("d"), Some("z"), Some("world"), Some("x"),]
1437 );
1438 }
1439
1440 #[test]
1441 fn test_struct_empty() {
1442 let fields = Fields::from(vec![
1443 Field::new("number_col", DataType::Int32, false),
1444 Field::new("string_col", DataType::Utf8, false),
1445 ]);
1446 let a = {
1447 let number_col = Int32Array::from_iter_values([1, 2, 3, 4]);
1448 let string_col = StringArray::from_iter_values(["a", "b", "c", "d"]);
1449
1450 StructArray::try_new(
1451 fields.clone(),
1452 vec![Arc::new(number_col), Arc::new(string_col)],
1453 None,
1454 )
1455 .unwrap()
1456 };
1457 let v = interleave(&[&a], &[]).unwrap();
1458 assert!(v.is_empty());
1459 assert_eq!(v.data_type(), &DataType::Struct(fields));
1460 }
1461
1462 #[test]
1463 fn interleave_sparse_nulls() {
1464 let values = StringArray::from_iter_values((0..100).map(|x| x.to_string()));
1465 let keys = Int32Array::from_iter_values(0..10);
1466 let dict_a = DictionaryArray::new(keys, Arc::new(values));
1467 let values = StringArray::new_null(0);
1468 let keys = Int32Array::new_null(10);
1469 let dict_b = DictionaryArray::new(keys, Arc::new(values));
1470
1471 let indices = &[(0, 0), (0, 1), (0, 2), (1, 0)];
1472 let array = interleave(&[&dict_a, &dict_b], indices).unwrap();
1473
1474 let expected =
1475 DictionaryArray::<Int32Type>::from_iter(vec![Some("0"), Some("1"), Some("2"), None]);
1476 assert_eq!(array.as_ref(), &expected)
1477 }
1478
1479 #[test]
1480 fn test_interleave_views() {
1481 let values = StringArray::from_iter_values([
1482 "hello",
1483 "world_long_string_not_inlined",
1484 "foo",
1485 "bar",
1486 "baz",
1487 ]);
1488 let view_a = StringViewArray::from(&values);
1489
1490 let values = StringArray::from_iter_values([
1491 "test",
1492 "data",
1493 "more_long_string_not_inlined",
1494 "views",
1495 "here",
1496 ]);
1497 let view_b = StringViewArray::from(&values);
1498
1499 let indices = &[
1500 (0, 2), (1, 0), (0, 4), (1, 3), (0, 1), ];
1506
1507 let values = interleave(&[&view_a, &view_b], indices).unwrap();
1509 let result = values.as_string_view();
1510 assert_eq!(result.data_buffers().len(), 1);
1511
1512 let fallback = interleave_fallback(&[&view_a, &view_b], indices).unwrap();
1513 let fallback_result = fallback.as_string_view();
1514 assert_eq!(fallback_result.data_buffers().len(), 2);
1516
1517 let collected: Vec<_> = result.iter().map(|x| x.map(|s| s.to_string())).collect();
1519
1520 let fallback_collected: Vec<_> = fallback_result
1521 .iter()
1522 .map(|x| x.map(|s| s.to_string()))
1523 .collect();
1524
1525 assert_eq!(&collected, &fallback_collected);
1526
1527 assert_eq!(
1528 &collected,
1529 &[
1530 Some("foo".to_string()),
1531 Some("test".to_string()),
1532 Some("baz".to_string()),
1533 Some("views".to_string()),
1534 Some("world_long_string_not_inlined".to_string()),
1535 ]
1536 );
1537 }
1538
1539 #[test]
1540 fn test_interleave_views_with_nulls() {
1541 let values = StringArray::from_iter([
1542 Some("hello"),
1543 None,
1544 Some("foo_long_string_not_inlined"),
1545 Some("bar"),
1546 None,
1547 ]);
1548 let view_a = StringViewArray::from(&values);
1549
1550 let values = StringArray::from_iter([
1551 Some("test"),
1552 Some("data_long_string_not_inlined"),
1553 None,
1554 None,
1555 Some("here"),
1556 ]);
1557 let view_b = StringViewArray::from(&values);
1558
1559 let indices = &[
1560 (0, 1), (1, 2), (0, 2), (1, 3), (0, 4), ];
1566
1567 let values = interleave(&[&view_a, &view_b], indices).unwrap();
1569 let result = values.as_string_view();
1570 assert_eq!(result.data_buffers().len(), 1);
1571
1572 let fallback = interleave_fallback(&[&view_a, &view_b], indices).unwrap();
1573 let fallback_result = fallback.as_string_view();
1574
1575 let collected: Vec<_> = result.iter().map(|x| x.map(|s| s.to_string())).collect();
1577
1578 let fallback_collected: Vec<_> = fallback_result
1579 .iter()
1580 .map(|x| x.map(|s| s.to_string()))
1581 .collect();
1582
1583 assert_eq!(&collected, &fallback_collected);
1584
1585 assert_eq!(
1586 &collected,
1587 &[
1588 None,
1589 None,
1590 Some("foo_long_string_not_inlined".to_string()),
1591 None,
1592 None,
1593 ]
1594 );
1595 }
1596
1597 #[test]
1598 fn test_interleave_views_multiple_buffers() {
1599 let str1 = "very_long_string_from_first_buffer".as_bytes();
1600 let str2 = "very_long_string_from_second_buffer".as_bytes();
1601 let buffer1 = str1.to_vec().into();
1602 let buffer2 = str2.to_vec().into();
1603
1604 let view1 = ByteView::new(str1.len() as u32, &str1[..4])
1605 .with_buffer_index(0)
1606 .with_offset(0)
1607 .as_u128();
1608 let view2 = ByteView::new(str2.len() as u32, &str2[..4])
1609 .with_buffer_index(1)
1610 .with_offset(0)
1611 .as_u128();
1612 let view_a =
1613 StringViewArray::try_new(vec![view1, view2].into(), vec![buffer1, buffer2], None)
1614 .unwrap();
1615
1616 let str3 = "another_very_long_string_buffer_three".as_bytes();
1617 let str4 = "different_long_string_in_buffer_four".as_bytes();
1618 let buffer3 = str3.to_vec().into();
1619 let buffer4 = str4.to_vec().into();
1620
1621 let view3 = ByteView::new(str3.len() as u32, &str3[..4])
1622 .with_buffer_index(0)
1623 .with_offset(0)
1624 .as_u128();
1625 let view4 = ByteView::new(str4.len() as u32, &str4[..4])
1626 .with_buffer_index(1)
1627 .with_offset(0)
1628 .as_u128();
1629 let view_b =
1630 StringViewArray::try_new(vec![view3, view4].into(), vec![buffer3, buffer4], None)
1631 .unwrap();
1632
1633 let indices = &[
1634 (0, 0), (1, 0), (0, 1), (1, 1), (0, 0), (1, 1), ];
1641
1642 let values = interleave(&[&view_a, &view_b], indices).unwrap();
1644 let result = values.as_string_view();
1645
1646 assert_eq!(
1647 result.data_buffers().len(),
1648 4,
1649 "Expected four buffers (two from each input array)"
1650 );
1651
1652 let result_strings: Vec<_> = result.iter().map(|x| x.map(|s| s.to_string())).collect();
1653 assert_eq!(
1654 result_strings,
1655 vec![
1656 Some("very_long_string_from_first_buffer".to_string()),
1657 Some("another_very_long_string_buffer_three".to_string()),
1658 Some("very_long_string_from_second_buffer".to_string()),
1659 Some("different_long_string_in_buffer_four".to_string()),
1660 Some("very_long_string_from_first_buffer".to_string()),
1661 Some("different_long_string_in_buffer_four".to_string()),
1662 ]
1663 );
1664
1665 let views = result.views();
1666 let buffer_indices: Vec<_> = views
1667 .iter()
1668 .map(|raw_view| ByteView::from(*raw_view).buffer_index)
1669 .collect();
1670
1671 assert_eq!(
1672 buffer_indices,
1673 vec![
1674 0, 1, 2, 3, 0, 3, ]
1681 );
1682 }
1683
1684 #[test]
1685 fn test_interleave_run_end_encoded_primitive() {
1686 let mut builder = PrimitiveRunBuilder::<Int32Type, Int32Type>::new();
1687 builder.extend([1, 1, 2, 2, 2, 3].into_iter().map(Some));
1688 let a = builder.finish();
1689
1690 let mut builder = PrimitiveRunBuilder::<Int32Type, Int32Type>::new();
1691 builder.extend([4, 5, 5, 6, 6, 6].into_iter().map(Some));
1692 let b = builder.finish();
1693
1694 let indices = &[(0, 1), (1, 0), (0, 4), (1, 2), (0, 5)];
1695 let result = interleave(&[&a, &b], indices).unwrap();
1696
1697 assert!(matches!(result.data_type(), DataType::RunEndEncoded(_, _)));
1699
1700 let result_run_array: &Int32RunArray = result.as_any().downcast_ref().unwrap();
1702
1703 let expected = vec![1, 4, 2, 5, 3];
1705 let mut actual = Vec::new();
1706 for i in 0..result_run_array.len() {
1707 let physical_idx = result_run_array.get_physical_index(i);
1708 let value = result_run_array
1709 .values()
1710 .as_primitive::<Int32Type>()
1711 .value(physical_idx);
1712 actual.push(value);
1713 }
1714 assert_eq!(actual, expected);
1715 }
1716
1717 #[test]
1718 fn test_interleave_run_end_encoded_sliced() {
1719 let mut builder = PrimitiveRunBuilder::<Int32Type, Int32Type>::new();
1720 builder.extend([1, 1, 2, 2, 2, 3].into_iter().map(Some));
1721 let a = builder.finish();
1722 let a = a.slice(2, 3); let mut builder = PrimitiveRunBuilder::<Int32Type, Int32Type>::new();
1725 builder.extend([4, 5, 5, 6, 6, 6].into_iter().map(Some));
1726 let b = builder.finish();
1727 let b = b.slice(1, 3); let indices = &[(0, 1), (1, 0), (0, 2), (1, 1), (1, 2)];
1730 let result = interleave(&[&a, &b], indices).unwrap();
1731
1732 let result = result.as_run::<Int32Type>();
1733 let result = result.downcast::<Int32Array>().unwrap();
1734
1735 let expected = vec![2, 5, 2, 5, 6];
1736 let actual = result.into_iter().flatten().collect::<Vec<_>>();
1737 assert_eq!(actual, expected);
1738 }
1739
1740 #[test]
1741 fn test_interleave_run_end_encoded_string() {
1742 let a: Int32RunArray = vec!["hello", "hello", "world", "world", "foo"]
1743 .into_iter()
1744 .collect();
1745 let b: Int32RunArray = vec!["bar", "baz", "baz", "qux"].into_iter().collect();
1746
1747 let indices = &[(0, 0), (1, 1), (0, 3), (1, 3), (0, 4)];
1748 let result = interleave(&[&a, &b], indices).unwrap();
1749
1750 assert!(matches!(result.data_type(), DataType::RunEndEncoded(_, _)));
1752
1753 let result_run_array: &Int32RunArray = result.as_any().downcast_ref().unwrap();
1755
1756 let expected = vec!["hello", "baz", "world", "qux", "foo"];
1758 let mut actual = Vec::new();
1759 for i in 0..result_run_array.len() {
1760 let physical_idx = result_run_array.get_physical_index(i);
1761 let value = result_run_array
1762 .values()
1763 .as_string::<i32>()
1764 .value(physical_idx);
1765 actual.push(value);
1766 }
1767 assert_eq!(actual, expected);
1768 }
1769
1770 #[test]
1771 fn test_interleave_run_end_encoded_with_nulls() {
1772 let a: Int32RunArray = vec![Some("a"), Some("a"), None, None, Some("b")]
1773 .into_iter()
1774 .collect();
1775 let b: Int32RunArray = vec![None, Some("c"), Some("c"), Some("d")]
1776 .into_iter()
1777 .collect();
1778
1779 let indices = &[(0, 1), (1, 0), (0, 2), (1, 3), (0, 4)];
1780 let result = interleave(&[&a, &b], indices).unwrap();
1781
1782 assert!(matches!(result.data_type(), DataType::RunEndEncoded(_, _)));
1784
1785 let result_run_array: &Int32RunArray = result.as_any().downcast_ref().unwrap();
1787
1788 let expected = vec![Some("a"), None, None, Some("d"), Some("b")];
1790 let mut actual = Vec::new();
1791 for i in 0..result_run_array.len() {
1792 let physical_idx = result_run_array.get_physical_index(i);
1793 if result_run_array.values().is_null(physical_idx) {
1794 actual.push(None);
1795 } else {
1796 let value = result_run_array
1797 .values()
1798 .as_string::<i32>()
1799 .value(physical_idx);
1800 actual.push(Some(value));
1801 }
1802 }
1803 assert_eq!(actual, expected);
1804 }
1805
1806 #[test]
1807 fn test_interleave_run_end_encoded_different_run_types() {
1808 let mut builder = PrimitiveRunBuilder::<Int16Type, Int32Type>::new();
1809 builder.extend([1, 1, 2, 3, 3].into_iter().map(Some));
1810 let a = builder.finish();
1811
1812 let mut builder = PrimitiveRunBuilder::<Int16Type, Int32Type>::new();
1813 builder.extend([4, 5, 5, 6].into_iter().map(Some));
1814 let b = builder.finish();
1815
1816 let indices = &[(0, 0), (1, 1), (0, 3), (1, 3)];
1817 let result = interleave(&[&a, &b], indices).unwrap();
1818
1819 assert!(matches!(result.data_type(), DataType::RunEndEncoded(_, _)));
1821
1822 let result_run_array: &RunArray<Int16Type> = result.as_any().downcast_ref().unwrap();
1824
1825 let expected = vec![1, 5, 3, 6];
1827 let mut actual = Vec::new();
1828 for i in 0..result_run_array.len() {
1829 let physical_idx = result_run_array.get_physical_index(i);
1830 let value = result_run_array
1831 .values()
1832 .as_primitive::<Int32Type>()
1833 .value(physical_idx);
1834 actual.push(value);
1835 }
1836 assert_eq!(actual, expected);
1837 }
1838
1839 #[test]
1840 fn test_interleave_run_end_encoded_mixed_run_lengths() {
1841 let mut builder = PrimitiveRunBuilder::<Int64Type, Int32Type>::new();
1842 builder.extend([1, 2, 2, 2, 2, 3, 3, 4].into_iter().map(Some));
1843 let a = builder.finish();
1844
1845 let mut builder = PrimitiveRunBuilder::<Int64Type, Int32Type>::new();
1846 builder.extend([5, 5, 5, 6, 7, 7, 8, 8].into_iter().map(Some));
1847 let b = builder.finish();
1848
1849 let indices = &[
1850 (0, 0), (1, 2), (0, 3), (1, 3), (0, 6), (1, 6), (0, 7), (1, 4), ];
1859 let result = interleave(&[&a, &b], indices).unwrap();
1860
1861 assert!(matches!(result.data_type(), DataType::RunEndEncoded(_, _)));
1863
1864 let result_run_array: &RunArray<Int64Type> = result.as_any().downcast_ref().unwrap();
1866
1867 let expected = vec![1, 5, 2, 6, 3, 8, 4, 7];
1869 let mut actual = Vec::new();
1870 for i in 0..result_run_array.len() {
1871 let physical_idx = result_run_array.get_physical_index(i);
1872 let value = result_run_array
1873 .values()
1874 .as_primitive::<Int32Type>()
1875 .value(physical_idx);
1876 actual.push(value);
1877 }
1878 assert_eq!(actual, expected);
1879 }
1880
1881 #[test]
1882 fn test_interleave_run_end_encoded_empty_runs() {
1883 let mut builder = PrimitiveRunBuilder::<Int32Type, Int32Type>::new();
1884 builder.extend([1].into_iter().map(Some));
1885 let a = builder.finish();
1886
1887 let mut builder = PrimitiveRunBuilder::<Int32Type, Int32Type>::new();
1888 builder.extend([2, 2, 2].into_iter().map(Some));
1889 let b = builder.finish();
1890
1891 let indices = &[(0, 0), (1, 1), (1, 2)];
1892 let result = interleave(&[&a, &b], indices).unwrap();
1893
1894 assert!(matches!(result.data_type(), DataType::RunEndEncoded(_, _)));
1896
1897 let result_run_array: &Int32RunArray = result.as_any().downcast_ref().unwrap();
1899
1900 let expected = vec![1, 2, 2];
1902 let mut actual = Vec::new();
1903 for i in 0..result_run_array.len() {
1904 let physical_idx = result_run_array.get_physical_index(i);
1905 let value = result_run_array
1906 .values()
1907 .as_primitive::<Int32Type>()
1908 .value(physical_idx);
1909 actual.push(value);
1910 }
1911 assert_eq!(actual, expected);
1912 }
1913
1914 #[test]
1915 fn test_struct_no_fields() {
1916 let fields = Fields::empty();
1917 let a = StructArray::try_new_with_length(fields.clone(), vec![], None, 10).unwrap();
1918 let v = interleave(&[&a], &[(0, 0)]).unwrap();
1919 assert_eq!(v.len(), 1);
1920 assert_eq!(v.data_type(), &DataType::Struct(fields));
1921 }
1922
1923 #[test]
1924 fn test_interleave_fallback_dictionary_with_nulls() {
1925 let input_1_keys = Int32Array::from_iter([Some(0), None, Some(1)]);
1926 let input_1_values = StringArray::from_iter_values(["foo", "bar"]);
1927 let dict_a = DictionaryArray::new(input_1_keys, Arc::new(input_1_values));
1928
1929 let input_2_keys = Int32Array::from_iter([Some(0), Some(1), None]);
1930 let input_2_values = StringArray::from_iter_values(["baz", "qux"]);
1931 let dict_b = DictionaryArray::new(input_2_keys, Arc::new(input_2_values));
1932
1933 let indices = vec![
1934 (0, 0), (0, 1), (1, 0), (1, 2), (0, 2), (1, 1), ];
1941
1942 let result =
1943 interleave_fallback_dictionary::<Int32Type>(&[&dict_a, &dict_b], &indices).unwrap();
1944 let dict_result = result.as_dictionary::<Int32Type>();
1945
1946 let string_result = dict_result.downcast_dict::<StringArray>().unwrap();
1947 let collected: Vec<_> = string_result.into_iter().collect();
1948 assert_eq!(
1949 collected,
1950 vec![
1951 Some("foo"),
1952 None,
1953 Some("baz"),
1954 None,
1955 Some("bar"),
1956 Some("qux")
1957 ]
1958 );
1959 }
1960
1961 #[test]
1962 fn test_interleave_bytes_offset_overflow() {
1963 let indices: Vec<(usize, usize)> = vec![(0, 0); (i32::MAX >> 4) as usize];
1964 let text = ('a'..='z').collect::<String>();
1965 let values = StringArray::from(vec![Some(text)]);
1966 assert!(matches!(
1967 interleave(&[&values], &indices),
1968 Err(ArrowError::OffsetOverflowError(_))
1969 ));
1970 }
1971
1972 #[test]
1973 fn test_interleave_list_offset_overflow() {
1974 let mut builder = GenericListBuilder::<i32, _>::new(Int32Builder::new());
1976 for i in 0..32 {
1977 builder.values().append_value(i);
1978 }
1979 builder.append(true);
1980 let list = builder.finish();
1981
1982 let indices: Vec<(usize, usize)> = vec![(0, 0); (i32::MAX as usize / 32) + 1];
1984 assert!(matches!(
1985 interleave(&[&list], &indices),
1986 Err(ArrowError::OffsetOverflowError(_))
1987 ));
1988 }
1989
1990 #[test]
1991 fn test_interleave_list_view() {
1992 let field = Arc::new(Field::new_list_field(DataType::Int64, false));
2001
2002 let lv_a = ListViewArray::new(
2003 Arc::clone(&field),
2004 ScalarBuffer::from(vec![0i32, 2]),
2005 ScalarBuffer::from(vec![2i32, 1]),
2006 Arc::new(Int64Array::from(vec![1_i64, 2, 3])),
2007 None,
2008 );
2009 let lv_b = ListViewArray::new(
2010 field,
2011 ScalarBuffer::from(vec![0i32]),
2012 ScalarBuffer::from(vec![3i32]),
2013 Arc::new(Int64Array::from(vec![4_i64, 5, 6])),
2014 None,
2015 );
2016
2017 let result = interleave(
2018 &[&lv_a as &dyn Array, &lv_b as &dyn Array],
2019 &[(0, 0), (1, 0), (0, 1)],
2020 )
2021 .unwrap();
2022
2023 result
2024 .to_data()
2025 .validate_full()
2026 .expect("interleaved ListViewArray must be internally consistent");
2027
2028 let result_lv = result.as_list_view::<i32>();
2029 assert_eq!(result_lv.len(), 3);
2030 assert_eq!(
2031 result_lv.value(0).as_primitive::<Int64Type>().values(),
2032 &[1, 2]
2033 );
2034 assert_eq!(
2035 result_lv.value(1).as_primitive::<Int64Type>().values(),
2036 &[4, 5, 6]
2037 );
2038 assert_eq!(
2039 result_lv.value(2).as_primitive::<Int64Type>().values(),
2040 &[3]
2041 );
2042 }
2043}