1use itertools::Itertools;
5use num_traits::AsPrimitive;
6use vortex_buffer::Buffer;
7use vortex_buffer::BufferMut;
8use vortex_error::VortexResult;
9use vortex_error::vortex_bail;
10use vortex_error::vortex_ensure;
11use vortex_error::vortex_err;
12use vortex_mask::Mask;
13
14use crate::ArrayRef;
15use crate::Canonical;
16use crate::Columnar;
17use crate::IntoArray;
18use crate::array::ArrayView;
19use crate::arrays::Chunked;
20use crate::arrays::ChunkedArray;
21use crate::arrays::ConstantArray;
22use crate::arrays::FixedSizeListArray;
23use crate::arrays::PiecewiseSequence;
24use crate::arrays::PiecewiseSequenceArray;
25use crate::arrays::PrimitiveArray;
26use crate::arrays::chunked::ChunkedArrayExt;
27use crate::arrays::dict::TakeExecute;
28use crate::arrays::piecewise_sequence::constant_unsigned_usize;
29use crate::arrays::piecewise_sequence::maybe_contiguous_slices;
30use crate::arrays::primitive::PrimitiveArrayExt;
31use crate::builders::ArrayBuilder;
32use crate::builders::builder_with_capacity;
33use crate::builtins::ArrayBuiltins;
34use crate::dtype::DType;
35use crate::dtype::PType;
36use crate::executor::ExecutionCtx;
37use crate::match_each_unsigned_integer_ptype;
38use crate::validity::Validity;
39
40enum ChunkFlattener {
44 Builder(Box<dyn ArrayBuilder>),
45 Chunks(Vec<ArrayRef>),
46}
47
48impl ChunkFlattener {
49 fn new(dtype: &DType, capacity: usize) -> Self {
50 if dtype.is_nested() {
51 Self::Chunks(Vec::new())
52 } else {
53 Self::Builder(builder_with_capacity(dtype, capacity))
54 }
55 }
56
57 fn push(&mut self, chunk: ArrayRef, ctx: &mut ExecutionCtx) -> VortexResult<()> {
58 match self {
59 Self::Builder(builder) => chunk.append_to_builder(builder.as_mut(), ctx),
60 Self::Chunks(chunks) => {
61 chunks.push(chunk);
62 Ok(())
63 }
64 }
65 }
66
67 fn finish(self, dtype: &DType, ctx: &mut ExecutionCtx) -> VortexResult<ArrayRef> {
68 match self {
69 Self::Builder(mut builder) => Ok(builder.finish()),
70 Self::Chunks(chunks) => unsafe { ChunkedArray::new_unchecked(chunks, dtype.clone()) }
73 .into_array()
74 .execute::<Canonical>(ctx)
75 .map(IntoArray::into_array),
76 }
77 }
78}
79
80fn take_chunked_via_sort(
81 array: ArrayView<'_, Chunked>,
82 indices: &PrimitiveArray,
83 indices_mask: Mask,
84 ctx: &mut ExecutionCtx,
85) -> VortexResult<ArrayRef> {
86 let indices_values = indices.as_slice::<u64>();
87 let n = indices_values.len();
88 let mut pairs: Vec<(u64, usize)> = indices_values
89 .iter()
90 .enumerate()
91 .filter(|&(position, _)| indices_mask.value(position))
92 .map(|(position, &index)| (index, position))
93 .collect();
94 pairs.sort_unstable();
95
96 if let Some(&(index, _)) = pairs.last() {
97 let index = usize::try_from(index)?;
98 if index >= array.len() {
99 vortex_bail!(OutOfBounds: index, 0, array.len());
100 }
101 }
102
103 let chunk_offsets = array.chunk_offset_values();
104 let nchunks = array.nchunks();
105 let mut flattener = ChunkFlattener::new(array.dtype(), pairs.len());
106 let mut final_take = BufferMut::<u64>::zeroed(n);
107 let mut cursor = 0usize;
108 let mut dedup_idx = 0u64;
109
110 for chunk_idx in 0..nchunks {
111 let chunk_start = chunk_offsets[chunk_idx];
112 let chunk_end = chunk_offsets[chunk_idx + 1];
113 let chunk_len = chunk_end - chunk_start;
114 let chunk_end_u64 = u64::try_from(chunk_end)?;
115 let range_end = cursor + pairs[cursor..].partition_point(|&(v, _)| v < chunk_end_u64);
116 let chunk_pairs = &pairs[cursor..range_end];
117
118 if !chunk_pairs.is_empty() {
119 let mut local_indices = Vec::new();
120 for (i, &(value, original_position)) in chunk_pairs.iter().enumerate() {
121 if cursor + i > 0 && value != pairs[cursor + i - 1].0 {
122 dedup_idx += 1;
123 }
124 let local_index = usize::try_from(value)? - chunk_start;
125 if local_indices.last() != Some(&local_index) {
126 local_indices.push(local_index);
127 }
128 final_take[original_position] = dedup_idx;
129 }
130
131 flattener.push(
132 array
133 .chunk(chunk_idx)
134 .filter(Mask::from_indices(chunk_len, local_indices))?,
135 ctx,
136 )?;
137 }
138
139 cursor = range_end;
140 }
141
142 let flat = flattener.finish(array.dtype(), ctx)?;
143 let take_validity = Validity::from_mask(indices_mask, indices.dtype().nullability());
144 flat.take(PrimitiveArray::new(final_take.freeze(), take_validity).into_array())
145}
146
147fn valid_indices_are_monotonic(indices: &[u64], indices_mask: &Mask) -> bool {
148 let mut previous = None;
149 for (position, &index) in indices.iter().enumerate() {
150 if !indices_mask.value(position) {
151 continue;
152 }
153 if previous.is_some_and(|previous| index < previous) {
154 return false;
155 }
156 previous = Some(index);
157 }
158 true
159}
160
161fn take_chunked(
163 array: ArrayView<'_, Chunked>,
164 indices: &ArrayRef,
165 ctx: &mut ExecutionCtx,
166) -> VortexResult<ArrayRef> {
167 let indices = indices
168 .cast(DType::Primitive(PType::U64, indices.dtype().nullability()))?
169 .execute::<PrimitiveArray>(ctx)?;
170
171 let indices_mask = indices
172 .as_ref()
173 .validity()?
174 .execute_mask(indices.as_ref().len(), ctx)?;
175 let indices_values = indices.as_slice::<u64>();
176 let n = indices_values.len();
177 let chunk_offsets = array.chunk_offset_values();
178 let nchunks = array.nchunks();
179
180 if !indices.as_ref().dtype().is_nullable() && indices_values.is_sorted_by(|a, b| a < b) {
183 return take_chunked_sorted(array, indices_values);
184 }
185
186 if matches!(array.dtype(), DType::FixedSizeList(..))
190 && n <= 64
191 && nchunks <= 16
192 && !valid_indices_are_monotonic(indices_values, &indices_mask)
193 {
194 return take_chunked_via_sort(array, &indices, indices_mask, ctx);
195 }
196
197 let mut buckets = vec![Vec::<(u64, usize)>::new(); nchunks];
200 let mut monotonic = true;
201 let mut last_index = None;
202 let mut sorted_chunk_idx = 0;
203
204 for (original_position, &index) in indices_values.iter().enumerate() {
205 if !indices_mask.value(original_position) {
206 continue;
207 }
208
209 let index = usize::try_from(index)?;
210 if index >= array.len() {
211 vortex_bail!(OutOfBounds: index, 0, array.len());
212 }
213
214 let still_sorted = last_index.is_none_or(|last_index| index >= last_index);
215 let chunk_idx = if monotonic && still_sorted {
216 while chunk_offsets[sorted_chunk_idx + 1] <= index {
217 sorted_chunk_idx += 1;
218 }
219 sorted_chunk_idx
220 } else {
221 monotonic = false;
222 chunk_offsets.partition_point(|&offset| offset <= index) - 1
223 };
224 last_index = Some(index);
225
226 let local_index = u64::try_from(index - chunk_offsets[chunk_idx])?;
227 buckets[chunk_idx].push((local_index, original_position));
228 }
229
230 let mut flattener = ChunkFlattener::new(array.dtype(), indices_mask.true_count());
231 let mut final_take =
232 (!monotonic || indices.dtype().is_nullable()).then(|| BufferMut::<u64>::zeroed(n));
233 let mut grouped_position = 0u64;
234
235 for (chunk_idx, bucket) in buckets.into_iter().enumerate() {
236 if bucket.is_empty() {
237 continue;
238 }
239
240 let mut local_indices = BufferMut::<u64>::with_capacity(bucket.len());
241 for (local_index, original_position) in bucket {
242 local_indices.push(local_index);
243 if let Some(final_take) = &mut final_take {
244 final_take[original_position] = grouped_position;
245 }
246 grouped_position += 1;
247 }
248
249 let local_indices =
250 PrimitiveArray::new(local_indices.freeze(), Validity::NonNullable).into_array();
251 flattener.push(array.chunk(chunk_idx).take(local_indices)?, ctx)?;
252 }
253
254 let flat = flattener.finish(array.dtype(), ctx)?;
256
257 let Some(final_take) = final_take else {
260 return Ok(flat);
261 };
262
263 let take_validity = Validity::from_mask(indices_mask, indices.dtype().nullability());
266 flat.take(PrimitiveArray::new(final_take.freeze(), take_validity).into_array())
267}
268
269fn take_chunked_sorted(
273 array: ArrayView<'_, Chunked>,
274 indices_values: &[u64],
275) -> VortexResult<ArrayRef> {
276 let chunk_offsets = array.chunk_offset_values();
277 let mut chunks = Vec::new();
278 let mut cursor = 0usize;
279
280 let first_chunk = match indices_values.first() {
283 Some(&first) => {
284 let first = usize::try_from(first)?;
285 chunk_offsets
286 .partition_point(|&offset| offset <= first)
287 .saturating_sub(1)
288 }
289 None => 0,
290 };
291
292 for chunk_idx in first_chunk..array.nchunks() {
293 if cursor == indices_values.len() {
294 break;
295 }
296 let chunk_start = chunk_offsets[chunk_idx];
297 let chunk_end = chunk_offsets[chunk_idx + 1];
298 let chunk_end_u64 = u64::try_from(chunk_end)?;
299
300 let range_end = cursor + indices_values[cursor..].partition_point(|&v| v < chunk_end_u64);
301 if range_end > cursor {
302 let chunk_start_u64 = u64::try_from(chunk_start)?;
303 let mut local_indices = BufferMut::<u64>::with_capacity(range_end - cursor);
304 for &val in &indices_values[cursor..range_end] {
305 local_indices.push(val - chunk_start_u64);
306 }
307 let local_indices =
308 PrimitiveArray::new(local_indices.freeze(), Validity::NonNullable).into_array();
309 chunks.push(array.chunk(chunk_idx).take(local_indices)?);
310 }
311 cursor = range_end;
312 }
313 if cursor != indices_values.len() {
314 vortex_bail!(
315 OutOfBounds: usize::try_from(indices_values[cursor])?, 0, array.as_ref().len()
316 );
317 }
318
319 if chunks.len() == 1 {
320 return Ok(chunks.swap_remove(0));
321 }
322 Ok(unsafe { ChunkedArray::new_unchecked(chunks, array.dtype().clone()) }.into_array())
325}
326
327impl TakeExecute for Chunked {
328 fn take(
329 array: ArrayView<'_, Chunked>,
330 indices: &ArrayRef,
331 ctx: &mut ExecutionCtx,
332 ) -> VortexResult<Option<ArrayRef>> {
333 if array.nchunks() == 1 {
334 return array.chunk(0).take(indices.clone()).map(Some);
335 }
336
337 if let Some(taken) = take_chunked_fsl(array, indices, ctx)? {
338 return Ok(Some(taken));
339 }
340
341 if let Some(piecewise_indices) = indices.as_opt::<PiecewiseSequence>()
342 && let Some(taken) = take_piecewise_chunked(array, piecewise_indices, ctx)?
343 {
344 return Ok(Some(taken));
345 }
346
347 take_chunked(array, indices, ctx).map(Some)
348 }
349}
350
351fn take_chunked_fsl(
356 array: ArrayView<'_, Chunked>,
357 indices: &ArrayRef,
358 ctx: &mut ExecutionCtx,
359) -> VortexResult<Option<ArrayRef>> {
360 if !matches!(array.dtype(), DType::FixedSizeList(..)) {
361 return Ok(None);
362 }
363
364 let fsl = array.as_ref().clone().execute::<FixedSizeListArray>(ctx)?;
365 fsl.into_array().take(indices.clone()).map(Some)
366}
367
368#[derive(Default)]
370struct ChunkGather {
371 starts: Vec<u64>,
372 lengths: Vec<u64>,
373 total: usize,
374}
375
376fn take_piecewise_chunked(
384 array: ArrayView<'_, Chunked>,
385 indices: ArrayView<'_, PiecewiseSequence>,
386 ctx: &mut ExecutionCtx,
387) -> VortexResult<Option<ArrayRef>> {
388 let Some((starts, lengths)) = maybe_contiguous_slices(indices, ctx)? else {
389 return Ok(None);
390 };
391
392 let output_len = indices.as_ref().len();
393 let starts: Vec<usize> = match_each_unsigned_integer_ptype!(starts.ptype(), |S| {
394 starts
395 .as_slice::<S>()
396 .iter()
397 .map(|&start| start.as_())
398 .collect()
399 });
400 let lengths: Vec<usize> = match lengths {
401 Columnar::Constant(lengths) => vec![constant_unsigned_usize(&lengths); starts.len()],
402 Columnar::Canonical(lengths) => {
403 let lengths = lengths.into_primitive();
404 match_each_unsigned_integer_ptype!(lengths.ptype(), |L| {
405 lengths
406 .as_slice::<L>()
407 .iter()
408 .map(|&length| length.as_())
409 .collect()
410 })
411 }
412 };
413
414 let array_len = array.as_ref().len();
415 let chunk_offsets = array.chunk_offset_values();
416 let nchunks = array.nchunks();
417
418 let mut plans: Vec<ChunkGather> = Vec::new();
419 plans.resize_with(nchunks, ChunkGather::default);
420 let mut sub_pieces: Vec<(usize, usize, usize)> = Vec::with_capacity(starts.len());
423 let mut monotonic = true;
424 let mut prev_chunk = 0usize;
425 let mut total_len = 0usize;
426
427 for (&start, &length) in starts.iter().zip_eq(&lengths) {
428 if length == 0 {
429 continue;
430 }
431 let end = start
432 .checked_add(length)
433 .ok_or_else(|| vortex_err!("PiecewiseSequenceArray range overflows usize"))?;
434 if end > array_len {
435 vortex_bail!(OutOfBounds: end - 1, 0, array_len);
436 }
437 total_len = total_len
438 .checked_add(length)
439 .ok_or_else(|| vortex_err!("PiecewiseSequenceArray output length overflows usize"))?;
440
441 let mut chunk_idx = chunk_offsets.partition_point(|&offset| offset <= start) - 1;
443 let mut cursor = start;
444 let mut remaining = length;
445 while remaining > 0 {
446 while chunk_offsets[chunk_idx + 1] <= cursor {
447 chunk_idx += 1;
448 }
449 let run = remaining.min(chunk_offsets[chunk_idx + 1] - cursor);
450 monotonic &= chunk_idx >= prev_chunk;
451 prev_chunk = chunk_idx;
452
453 let plan = &mut plans[chunk_idx];
454 match sub_pieces.last_mut() {
457 Some((last_chunk, _, last_run)) if *last_chunk == chunk_idx => *last_run += run,
458 _ => sub_pieces.push((chunk_idx, plan.total, run)),
459 }
460 let local_start = (cursor - chunk_offsets[chunk_idx]) as u64;
463 match (plan.starts.last(), plan.lengths.last_mut()) {
464 (Some(&prev_start), Some(prev_len)) if prev_start + *prev_len == local_start => {
465 *prev_len += run as u64;
466 }
467 _ => {
468 plan.starts.push(local_start);
469 plan.lengths.push(run as u64);
470 }
471 }
472 plan.total += run;
473
474 cursor += run;
475 remaining -= run;
476 }
477 }
478
479 vortex_ensure!(
480 total_len == output_len,
481 "PiecewiseSequenceArray expanded length {total_len} does not match declared length {output_len}"
482 );
483
484 if monotonic {
486 let mut gathered = Vec::new();
487 for (chunk_idx, plan) in plans.into_iter().enumerate() {
488 if plan.starts.is_empty() {
489 continue;
490 }
491 gathered.push(gather_chunk(array.chunk(chunk_idx), plan)?);
492 }
493 let result = if gathered.len() == 1 {
494 gathered.swap_remove(0)
495 } else {
496 unsafe { ChunkedArray::new_unchecked(gathered, array.dtype().clone()) }.into_array()
499 };
500 return Ok(Some(result));
501 }
502
503 let mut bases = vec![0usize; nchunks];
506 let mut running = 0usize;
507 let mut flattener = ChunkFlattener::new(array.dtype(), output_len);
508 for (chunk_idx, plan) in plans.into_iter().enumerate() {
509 if plan.starts.is_empty() {
510 continue;
511 }
512 bases[chunk_idx] = running;
513 running += plan.total;
514 flattener.push(gather_chunk(array.chunk(chunk_idx), plan)?, ctx)?;
515 }
516 let flat = flattener.finish(array.dtype(), ctx)?;
517
518 let mut reorder_starts = Vec::with_capacity(sub_pieces.len());
519 let mut reorder_lengths = Vec::with_capacity(sub_pieces.len());
520 for &(chunk_idx, offset, run) in &sub_pieces {
521 reorder_starts.push((bases[chunk_idx] + offset) as u64);
522 reorder_lengths.push(run as u64);
523 }
524 flat.take(contiguous_runs(reorder_starts, reorder_lengths, output_len))
525 .map(Some)
526}
527
528fn gather_chunk(chunk: &ArrayRef, plan: ChunkGather) -> VortexResult<ArrayRef> {
531 if let [start] = plan.starts.as_slice() {
532 let start = usize::try_from(*start)?;
533 return chunk.slice(start..start + plan.total);
534 }
535 chunk.take(contiguous_runs(plan.starts, plan.lengths, plan.total))
536}
537
538fn contiguous_runs(starts: Vec<u64>, lengths: Vec<u64>, total: usize) -> ArrayRef {
541 let count = starts.len();
542 debug_assert_eq!(count, lengths.len());
543 let starts = PrimitiveArray::new(Buffer::from(starts), Validity::NonNullable).into_array();
544 let lengths = match lengths.first() {
545 Some(&first) if lengths.iter().all(|&length| length == first) => {
546 ConstantArray::new(first, count).into_array()
547 }
548 _ => PrimitiveArray::new(Buffer::from(lengths), Validity::NonNullable).into_array(),
549 };
550 let multipliers = ConstantArray::new(1u64, count).into_array();
551 unsafe { PiecewiseSequenceArray::new_unchecked(starts, lengths, multipliers, total) }
554 .into_array()
555}
556
557#[cfg(test)]
558mod tests {
559 use vortex_buffer::Buffer;
560 use vortex_buffer::bitbuffer;
561 use vortex_buffer::buffer;
562 use vortex_error::VortexResult;
563
564 use crate::ArrayRef;
565 use crate::Canonical;
566 use crate::IntoArray;
567 use crate::VortexSessionExecute;
568 use crate::array_session;
569 use crate::arrays::BoolArray;
570 use crate::arrays::ChunkedArray;
571 use crate::arrays::ConstantArray;
572 use crate::arrays::FixedSizeListArray;
573 use crate::arrays::PiecewiseSequenceArray;
574 use crate::arrays::PrimitiveArray;
575 use crate::arrays::StructArray;
576 use crate::arrays::chunked::ChunkedArrayExt;
577 use crate::assert_arrays_eq;
578 use crate::compute::conformance::take::test_take_conformance;
579 use crate::dtype::DType;
580 use crate::dtype::FieldNames;
581 use crate::dtype::Nullability;
582 use crate::dtype::PType;
583 use crate::scalar::Scalar;
584 use crate::validity::Validity;
585
586 fn chunked_i32() -> VortexResult<ChunkedArray> {
587 ChunkedArray::try_new(
588 vec![
589 buffer![0i32, 1, 2, 3, 4].into_array(),
590 buffer![5i32, 6, 7, 8, 9].into_array(),
591 buffer![10i32, 11, 12, 13, 14].into_array(),
592 ],
593 DType::Primitive(PType::I32, Nullability::NonNullable),
594 )
595 }
596
597 fn contiguous_pieces(starts: &[u64], lengths: &[u64]) -> VortexResult<ArrayRef> {
598 let len = usize::try_from(lengths.iter().sum::<u64>())?;
599 Ok(PiecewiseSequenceArray::try_new(
600 starts.iter().copied().collect::<Buffer<u64>>().into_array(),
601 lengths
602 .iter()
603 .copied()
604 .collect::<Buffer<u64>>()
605 .into_array(),
606 ConstantArray::new(1u64, starts.len()).into_array(),
607 len,
608 )?
609 .into_array())
610 }
611
612 #[test]
613 fn test_take_piecewise_monotonic_spanning_chunks() -> VortexResult<()> {
614 let mut ctx = array_session().create_execution_ctx();
615 let arr = chunked_i32()?;
616
617 let indices = contiguous_pieces(&[1, 4, 9], &[3, 4, 6])?;
619 let result = arr.take(indices)?;
620
621 assert_arrays_eq!(
622 result,
623 PrimitiveArray::from_iter([1i32, 2, 3, 4, 5, 6, 7, 9, 10, 11, 12, 13, 14]),
624 &mut ctx
625 );
626 Ok(())
627 }
628
629 #[test]
630 fn test_take_piecewise_interleaved() -> VortexResult<()> {
631 let mut ctx = array_session().create_execution_ctx();
632 let arr = chunked_i32()?;
633
634 let indices = contiguous_pieces(&[12, 2, 7, 0], &[3, 2, 3, 1])?;
636 let result = arr.take(indices)?;
637
638 assert_arrays_eq!(
639 result,
640 PrimitiveArray::from_iter([12i32, 13, 14, 2, 3, 7, 8, 9, 0]),
641 &mut ctx
642 );
643 Ok(())
644 }
645
646 #[test]
647 fn test_take_piecewise_whole_array() -> VortexResult<()> {
648 let mut ctx = array_session().create_execution_ctx();
649 let arr = chunked_i32()?;
650
651 let indices = contiguous_pieces(&[0], &[15])?;
652 let result = arr.take(indices)?;
653
654 assert_arrays_eq!(result, PrimitiveArray::from_iter(0i32..15), &mut ctx);
655 Ok(())
656 }
657
658 #[test]
659 fn test_take_piecewise_across_empty_chunk() -> VortexResult<()> {
660 let mut ctx = array_session().create_execution_ctx();
661 let arr = ChunkedArray::try_new(
662 vec![
663 buffer![0i32, 1, 2, 3, 4].into_array(),
664 PrimitiveArray::empty::<i32>(Nullability::NonNullable).into_array(),
665 buffer![5i32, 6, 7, 8, 9].into_array(),
666 ],
667 DType::Primitive(PType::I32, Nullability::NonNullable),
668 )?;
669
670 let indices = contiguous_pieces(&[3], &[4])?;
671 let result = arr.take(indices)?;
672
673 assert_arrays_eq!(result, PrimitiveArray::from_iter([3i32, 4, 5, 6]), &mut ctx);
674 Ok(())
675 }
676
677 #[test]
678 fn test_take_piecewise_adjacent_pieces_merge() -> VortexResult<()> {
679 let mut ctx = array_session().create_execution_ctx();
680 let arr = chunked_i32()?;
681
682 let indices = contiguous_pieces(&[0, 2, 7], &[2, 2, 2])?;
685 let result = arr.take(indices)?;
686
687 assert_arrays_eq!(
688 result,
689 PrimitiveArray::from_iter([0i32, 1, 2, 3, 7, 8]),
690 &mut ctx
691 );
692 Ok(())
693 }
694
695 #[test]
696 fn test_take_piecewise_same_chunk_out_of_order() -> VortexResult<()> {
697 let mut ctx = array_session().create_execution_ctx();
698 let arr = chunked_i32()?;
699
700 let indices = contiguous_pieces(&[2, 0], &[2, 2])?;
702 let result = arr.take(indices)?;
703
704 assert_arrays_eq!(result, PrimitiveArray::from_iter([2i32, 3, 0, 1]), &mut ctx);
705 Ok(())
706 }
707
708 #[test]
709 fn test_take_piecewise_zero_length_pieces() -> VortexResult<()> {
710 let mut ctx = array_session().create_execution_ctx();
711 let arr = chunked_i32()?;
712
713 let indices = contiguous_pieces(&[5, 0, 2], &[0, 4, 0])?;
714 let result = arr.take(indices)?;
715
716 assert_arrays_eq!(result, PrimitiveArray::from_iter([0i32, 1, 2, 3]), &mut ctx);
717 Ok(())
718 }
719
720 #[test]
721 fn test_take_piecewise_out_of_bounds() -> VortexResult<()> {
722 let mut ctx = array_session().create_execution_ctx();
723 let arr = chunked_i32()?;
724
725 let indices = contiguous_pieces(&[12], &[5])?;
726 let result = arr
727 .take(indices)
728 .and_then(|taken| taken.execute::<Canonical>(&mut ctx));
729
730 assert!(result.is_err());
731 Ok(())
732 }
733
734 #[test]
735 fn test_take_piecewise_non_unit_multiplier() -> VortexResult<()> {
736 let mut ctx = array_session().create_execution_ctx();
737 let arr = chunked_i32()?;
738
739 let indices = PiecewiseSequenceArray::try_new(
741 buffer![0u64, 1].into_array(),
742 buffer![5u64, 3].into_array(),
743 buffer![2u64, 4].into_array(),
744 8,
745 )?
746 .into_array();
747 let result = arr.take(indices)?;
748
749 assert_arrays_eq!(
750 result,
751 PrimitiveArray::from_iter([0i32, 2, 4, 6, 8, 1, 5, 9]),
752 &mut ctx
753 );
754 Ok(())
755 }
756
757 #[test]
758 fn test_take_piecewise_nullable_values() -> VortexResult<()> {
759 let mut ctx = array_session().create_execution_ctx();
760 let arr = ChunkedArray::try_new(
761 vec![
762 PrimitiveArray::from_option_iter([Some(0i32), None, Some(2)]).into_array(),
763 PrimitiveArray::from_option_iter([None, Some(4i32), Some(5)]).into_array(),
764 ],
765 DType::Primitive(PType::I32, Nullability::Nullable),
766 )?;
767
768 let indices = contiguous_pieces(&[4, 1], &[2, 3])?;
769 let result = arr.take(indices)?;
770
771 assert_arrays_eq!(
772 result,
773 PrimitiveArray::from_option_iter([Some(4i32), Some(5), None, Some(2), None]),
774 &mut ctx
775 );
776 Ok(())
777 }
778
779 #[test]
780 fn test_take_fsl_over_chunked_elements() -> VortexResult<()> {
781 let mut ctx = array_session().create_execution_ctx();
782 let elements = ChunkedArray::try_new(
784 vec![
785 PrimitiveArray::from_iter(0i32..8).into_array(),
786 PrimitiveArray::from_iter(8i32..13).into_array(),
787 PrimitiveArray::from_iter(13i32..18).into_array(),
788 ],
789 DType::Primitive(PType::I32, Nullability::NonNullable),
790 )?
791 .into_array();
792 let fsl = FixedSizeListArray::try_new(elements, 3, Validity::NonNullable, 6)?;
793
794 let indices = buffer![4u64, 0, 2, 5, 1, 4].into_array();
795 let result = fsl.take(indices.clone())?;
796 let expected = FixedSizeListArray::try_new(
797 PrimitiveArray::from_iter(0i32..18).into_array(),
798 3,
799 Validity::NonNullable,
800 6,
801 )?
802 .take(indices)?;
803
804 assert_arrays_eq!(result, expected, &mut ctx);
805 Ok(())
806 }
807
808 #[test]
809 fn test_take_fsl_over_chunked_elements_conformance() -> VortexResult<()> {
810 let elements = ChunkedArray::try_new(
811 vec![
812 PrimitiveArray::from_iter(0i32..8).into_array(),
813 PrimitiveArray::from_iter(8i32..13).into_array(),
814 PrimitiveArray::from_iter(13i32..18).into_array(),
815 ],
816 DType::Primitive(PType::I32, Nullability::NonNullable),
817 )?
818 .into_array();
819 let fsl = FixedSizeListArray::try_new(elements, 3, Validity::NonNullable, 6)?;
820 test_take_conformance(
821 &fsl.into_array(),
822 &mut array_session().create_execution_ctx(),
823 );
824 Ok(())
825 }
826
827 #[test]
828 fn test_take_chunked_fsl() -> VortexResult<()> {
829 let mut ctx = array_session().create_execution_ctx();
830 let c0 = FixedSizeListArray::try_new(
831 PrimitiveArray::from_iter(0i32..6).into_array(),
832 2,
833 Validity::NonNullable,
834 3,
835 )?;
836 let c1 = FixedSizeListArray::try_new(
837 PrimitiveArray::from_iter(6i32..10).into_array(),
838 2,
839 Validity::NonNullable,
840 2,
841 )?;
842 let dtype = c0.dtype().clone();
843 let arr = ChunkedArray::try_new(vec![c0.into_array(), c1.into_array()], dtype)?;
844
845 let indices = buffer![4u64, 0, 3, 3, 1].into_array();
846 let result = arr.take(indices.clone())?;
847 let expected = FixedSizeListArray::try_new(
848 PrimitiveArray::from_iter(0i32..10).into_array(),
849 2,
850 Validity::NonNullable,
851 5,
852 )?
853 .take(indices)?;
854
855 assert_arrays_eq!(result, expected, &mut ctx);
856 Ok(())
857 }
858
859 #[test]
860 fn test_take_chunked_fsl_nullable() -> VortexResult<()> {
861 let mut ctx = array_session().create_execution_ctx();
862 let c0 = FixedSizeListArray::try_new(
863 PrimitiveArray::from_iter(0i32..6).into_array(),
864 2,
865 Validity::Array(bitbuffer![1 0 1].into_array()),
866 3,
867 )?;
868 let c1 = FixedSizeListArray::try_new(
869 PrimitiveArray::from_iter(6i32..10).into_array(),
870 2,
871 Validity::AllValid,
872 2,
873 )?;
874 let dtype = c0.dtype().clone();
875 let arr = ChunkedArray::try_new(vec![c0.into_array(), c1.into_array()], dtype)?;
876
877 let indices =
878 PrimitiveArray::from_option_iter([Some(4u64), None, Some(0), Some(1)]).into_array();
879 let result = arr.take(indices.clone())?;
880 let expected = FixedSizeListArray::try_new(
881 PrimitiveArray::from_iter(0i32..10).into_array(),
882 2,
883 Validity::Array(bitbuffer![1 0 1 1 1].into_array()),
884 5,
885 )?
886 .take(indices)?;
887
888 assert_arrays_eq!(result, expected, &mut ctx);
889 Ok(())
890 }
891
892 #[test]
893 fn test_take_chunked_fsl_non_fsl_chunk() -> VortexResult<()> {
894 let mut ctx = array_session().create_execution_ctx();
895 let c0 = FixedSizeListArray::try_new(
896 PrimitiveArray::from_iter(0i32..6).into_array(),
897 2,
898 Validity::NonNullable,
899 3,
900 )?;
901 let c1 = FixedSizeListArray::try_new(
902 PrimitiveArray::from_iter(6i32..10).into_array(),
903 2,
904 Validity::NonNullable,
905 2,
906 )?;
907 let dtype = c0.dtype().clone();
908 let nested = ChunkedArray::try_new(vec![c1.into_array()], dtype.clone())?;
911 let arr = ChunkedArray::try_new(vec![c0.into_array(), nested.into_array()], dtype)?;
912
913 let indices = buffer![4u64, 0, 3, 1].into_array();
914 let result = arr.take(indices.clone())?;
915 let expected = FixedSizeListArray::try_new(
916 PrimitiveArray::from_iter(0i32..10).into_array(),
917 2,
918 Validity::NonNullable,
919 5,
920 )?
921 .take(indices)?;
922
923 assert_arrays_eq!(result, expected, &mut ctx);
924 Ok(())
925 }
926
927 #[test]
928 fn test_take_chunked_fsl_constant_chunk() -> VortexResult<()> {
929 let mut ctx = array_session().create_execution_ctx();
930 let c0 = FixedSizeListArray::try_new(
931 PrimitiveArray::from_iter(0i32..6).into_array(),
932 2,
933 Validity::AllValid,
934 3,
935 )?;
936 let dtype = c0.dtype().clone();
937 let c1 = ConstantArray::new(Scalar::null(dtype.clone()), 2).into_array();
939 let arr = ChunkedArray::try_new(vec![c0.into_array(), c1], dtype)?;
940
941 let indices = buffer![4u64, 0, 3, 1].into_array();
942 let result = arr.take(indices.clone())?;
943 let expected = FixedSizeListArray::try_new(
944 PrimitiveArray::from_iter(0i32..10).into_array(),
945 2,
946 Validity::Array(bitbuffer![1 1 1 0 0].into_array()),
947 5,
948 )?
949 .take(indices)?;
950
951 assert_arrays_eq!(result, expected, &mut ctx);
952 Ok(())
953 }
954
955 #[test]
956 fn test_take_piecewise_chunked_fsl() -> VortexResult<()> {
957 let mut ctx = array_session().create_execution_ctx();
958 let c0 = FixedSizeListArray::try_new(
959 PrimitiveArray::from_iter(0i32..6).into_array(),
960 2,
961 Validity::NonNullable,
962 3,
963 )?;
964 let c1 = FixedSizeListArray::try_new(
965 PrimitiveArray::from_iter(6i32..10).into_array(),
966 2,
967 Validity::NonNullable,
968 2,
969 )?;
970 let dtype = c0.dtype().clone();
971 let arr = ChunkedArray::try_new(vec![c0.into_array(), c1.into_array()], dtype)?;
972 let reference = FixedSizeListArray::try_new(
973 PrimitiveArray::from_iter(0i32..10).into_array(),
974 2,
975 Validity::NonNullable,
976 5,
977 )?;
978
979 let indices = contiguous_pieces(&[1], &[3])?;
981 assert_arrays_eq!(
982 arr.take(indices.clone())?,
983 reference.take(indices)?,
984 &mut ctx
985 );
986
987 let indices = contiguous_pieces(&[3, 0], &[2, 2])?;
989 assert_arrays_eq!(
990 arr.take(indices.clone())?,
991 reference.take(indices)?,
992 &mut ctx
993 );
994 Ok(())
995 }
996
997 #[test]
998 fn test_take_chunked_struct_nested_flatten() -> VortexResult<()> {
999 let mut ctx = array_session().create_execution_ctx();
1000 let s0 = StructArray::try_new(
1001 ["a"].into(),
1002 vec![buffer![0i32, 1, 2].into_array()],
1003 3,
1004 Validity::NonNullable,
1005 )?;
1006 let s1 = StructArray::try_new(
1007 ["a"].into(),
1008 vec![buffer![3i32, 4].into_array()],
1009 2,
1010 Validity::NonNullable,
1011 )?;
1012 let dtype = s0.dtype().clone();
1013 let arr = ChunkedArray::try_new(vec![s0.into_array(), s1.into_array()], dtype)?;
1014
1015 let result = arr.take(buffer![4u64, 0, 2, 4].into_array())?;
1016 let expected = StructArray::try_new(
1017 ["a"].into(),
1018 vec![buffer![4i32, 0, 2, 4].into_array()],
1019 4,
1020 Validity::NonNullable,
1021 )?;
1022
1023 assert_arrays_eq!(result, expected, &mut ctx);
1024 Ok(())
1025 }
1026
1027 #[test]
1028 fn test_take_chunked_fsl_conformance() -> VortexResult<()> {
1029 let c0 = FixedSizeListArray::try_new(
1030 PrimitiveArray::from_iter(0i32..6).into_array(),
1031 2,
1032 Validity::NonNullable,
1033 3,
1034 )?;
1035 let c1 = FixedSizeListArray::try_new(
1036 PrimitiveArray::from_iter(6i32..10).into_array(),
1037 2,
1038 Validity::NonNullable,
1039 2,
1040 )?;
1041 let dtype = c0.dtype().clone();
1042 let arr = ChunkedArray::try_new(vec![c0.into_array(), c1.into_array()], dtype)?;
1043 test_take_conformance(
1044 &arr.into_array(),
1045 &mut array_session().create_execution_ctx(),
1046 );
1047 Ok(())
1048 }
1049
1050 #[test]
1051 fn test_take() {
1052 let mut ctx = array_session().create_execution_ctx();
1053 let a = buffer![1i32, 2, 3].into_array();
1054 let arr = ChunkedArray::try_new(vec![a.clone(), a.clone(), a.clone()], a.dtype().clone())
1055 .unwrap();
1056 assert_eq!(arr.nchunks(), 3);
1057 assert_eq!(arr.len(), 9);
1058 let indices = buffer![0u64, 0, 6, 4].into_array();
1059
1060 let result = arr.take(indices).unwrap();
1061 assert_arrays_eq!(result, PrimitiveArray::from_iter([1i32, 1, 1, 2]), &mut ctx);
1062 }
1063
1064 #[test]
1065 fn test_take_nullable_values() {
1066 let mut ctx = array_session().create_execution_ctx();
1067 let a = PrimitiveArray::new(buffer![1i32, 2, 3], Validity::AllValid).into_array();
1068 let arr = ChunkedArray::try_new(vec![a.clone(), a.clone(), a.clone()], a.dtype().clone())
1069 .unwrap();
1070 assert_eq!(arr.nchunks(), 3);
1071 assert_eq!(arr.len(), 9);
1072 let indices = PrimitiveArray::new(buffer![0u64, 0, 6, 4], Validity::NonNullable);
1073
1074 let result = arr.take(indices.into_array()).unwrap();
1075 assert_arrays_eq!(
1076 result,
1077 PrimitiveArray::from_option_iter([1i32, 1, 1, 2].map(Some)),
1078 &mut ctx
1079 );
1080 }
1081
1082 #[test]
1083 fn test_take_nullable_indices() {
1084 let mut ctx = array_session().create_execution_ctx();
1085 let a = buffer![1i32, 2, 3].into_array();
1086 let arr = ChunkedArray::try_new(vec![a.clone(), a.clone(), a.clone()], a.dtype().clone())
1087 .unwrap();
1088 assert_eq!(arr.nchunks(), 3);
1089 assert_eq!(arr.len(), 9);
1090 let indices = PrimitiveArray::new(
1091 buffer![0u64, 0, 6, 4],
1092 Validity::Array(bitbuffer![1 0 0 1].into_array()),
1093 );
1094
1095 let result = arr.take(indices.into_array()).unwrap();
1096 assert_arrays_eq!(
1097 result,
1098 PrimitiveArray::from_option_iter([Some(1i32), None, None, Some(2)]),
1099 &mut ctx
1100 );
1101 }
1102
1103 #[test]
1104 fn test_take_nullable_struct() {
1105 let mut ctx = array_session().create_execution_ctx();
1106 let struct_array =
1107 StructArray::try_new(FieldNames::default(), vec![], 100, Validity::NonNullable)
1108 .unwrap();
1109
1110 let arr = ChunkedArray::from_iter(vec![
1111 struct_array.clone().into_array(),
1112 struct_array.into_array(),
1113 ]);
1114
1115 let result = arr
1116 .take(PrimitiveArray::from_option_iter(vec![Some(0), None, Some(101)]).into_array())
1117 .unwrap();
1118
1119 let expect = StructArray::try_new(
1120 FieldNames::default(),
1121 vec![],
1122 3,
1123 Validity::Array(BoolArray::from_iter(vec![true, false, true]).into_array()),
1124 )
1125 .unwrap();
1126 assert_arrays_eq!(result, expect, &mut ctx);
1127 }
1128
1129 #[test]
1130 fn test_empty_take() {
1131 let mut ctx = array_session().create_execution_ctx();
1132 let a = buffer![1i32, 2, 3].into_array();
1133 let arr = ChunkedArray::try_new(vec![a.clone(), a.clone(), a.clone()], a.dtype().clone())
1134 .unwrap();
1135 assert_eq!(arr.nchunks(), 3);
1136 assert_eq!(arr.len(), 9);
1137
1138 let indices = PrimitiveArray::empty::<u64>(Nullability::NonNullable);
1139 let result = arr.take(indices.into_array()).unwrap();
1140
1141 assert!(result.is_empty());
1142 assert_eq!(result.dtype(), arr.dtype());
1143 assert_arrays_eq!(
1144 result,
1145 PrimitiveArray::empty::<i32>(Nullability::NonNullable),
1146 &mut ctx
1147 );
1148 }
1149
1150 #[test]
1151 fn test_take_shuffled_indices() -> VortexResult<()> {
1152 let mut ctx = array_session().create_execution_ctx();
1153 let c0 = buffer![0i32, 1, 2].into_array();
1154 let c1 = buffer![3i32, 4, 5].into_array();
1155 let c2 = buffer![6i32, 7, 8].into_array();
1156 let arr = ChunkedArray::try_new(
1157 vec![c0, c1, c2],
1158 PrimitiveArray::empty::<i32>(Nullability::NonNullable)
1159 .dtype()
1160 .clone(),
1161 )?;
1162
1163 let indices = buffer![8u64, 0, 5, 3, 2, 7, 1, 6, 4].into_array();
1165 let result = arr.take(indices)?;
1166
1167 assert_arrays_eq!(
1168 result,
1169 PrimitiveArray::from_iter([8i32, 0, 5, 3, 2, 7, 1, 6, 4]),
1170 &mut ctx
1171 );
1172 Ok(())
1173 }
1174
1175 #[test]
1176 fn test_take_shuffled_duplicates_with_empty_chunks() -> VortexResult<()> {
1177 let mut ctx = array_session().create_execution_ctx();
1178 let empty = PrimitiveArray::empty::<i32>(Nullability::NonNullable).into_array();
1179 let arr = ChunkedArray::try_new(
1180 vec![
1181 empty.clone(),
1182 buffer![0i32, 1].into_array(),
1183 empty.clone(),
1184 buffer![2i32, 3].into_array(),
1185 empty,
1186 ],
1187 PrimitiveArray::empty::<i32>(Nullability::NonNullable)
1188 .dtype()
1189 .clone(),
1190 )?;
1191
1192 let result = arr.take(buffer![3u64, 0, 2, 0, 3, 1, 2].into_array())?;
1193
1194 assert_arrays_eq!(
1195 result,
1196 PrimitiveArray::from_iter([3i32, 0, 2, 0, 3, 1, 2]),
1197 &mut ctx
1198 );
1199 Ok(())
1200 }
1201
1202 #[test]
1203 fn test_take_small_shuffled_fixed_size_lists() -> VortexResult<()> {
1204 let mut ctx = array_session().create_execution_ctx();
1205 let first = FixedSizeListArray::new(
1206 buffer![0i32, 1, 2, 3].into_array(),
1207 2,
1208 Validity::NonNullable,
1209 2,
1210 )
1211 .into_array();
1212 let second = FixedSizeListArray::new(
1213 buffer![4i32, 5, 6, 7].into_array(),
1214 2,
1215 Validity::NonNullable,
1216 2,
1217 )
1218 .into_array();
1219 let dtype = first.dtype().clone();
1220 let array = ChunkedArray::try_new(vec![first, second], dtype)?;
1221
1222 let result = array.take(buffer![3u64, 0, 2, 1, 3].into_array())?;
1223 let expected = FixedSizeListArray::new(
1224 buffer![6i32, 7, 0, 1, 4, 5, 2, 3, 6, 7].into_array(),
1225 2,
1226 Validity::NonNullable,
1227 5,
1228 );
1229
1230 assert_arrays_eq!(result, expected, &mut ctx);
1231 Ok(())
1232 }
1233
1234 #[test]
1235 fn test_take_shuffled_large() -> VortexResult<()> {
1236 let mut ctx = array_session().create_execution_ctx();
1237 let nchunks: i32 = 100;
1238 let chunk_len: i32 = 1_000;
1239 let total = nchunks * chunk_len;
1240
1241 let chunks: Vec<_> = (0..nchunks)
1242 .map(|c| {
1243 let start = c * chunk_len;
1244 PrimitiveArray::from_iter(start..start + chunk_len).into_array()
1245 })
1246 .collect();
1247 let dtype = chunks[0].dtype().clone();
1248 let arr = ChunkedArray::try_new(chunks, dtype)?;
1249
1250 let mut indices: Vec<u64> = (0..u64::try_from(total)?).collect();
1252 let mut seed: u64 = 0xdeadbeef;
1253 for i in (1..indices.len()).rev() {
1254 seed = seed.wrapping_mul(6364136223846793005).wrapping_add(1);
1255 let j = (seed >> 33) as usize % (i + 1);
1256 indices.swap(i, j);
1257 }
1258
1259 let indices_arr = PrimitiveArray::new(Buffer::from(indices.clone()), Validity::NonNullable);
1260 let result = arr.take(indices_arr.into_array())?;
1261
1262 let result = result.execute::<PrimitiveArray>(&mut ctx)?;
1264 let result_vals = result.as_slice::<i32>();
1265 for (pos, &idx) in indices.iter().enumerate() {
1266 assert_eq!(
1267 result_vals[pos],
1268 i32::try_from(idx)?,
1269 "mismatch at position {pos}"
1270 );
1271 }
1272 Ok(())
1273 }
1274
1275 #[test]
1276 fn test_take_null_indices() -> VortexResult<()> {
1277 let mut ctx = array_session().create_execution_ctx();
1278 let c0 = buffer![10i32, 20, 30].into_array();
1279 let c1 = buffer![40i32, 50, 60].into_array();
1280 let arr = ChunkedArray::try_new(
1281 vec![c0, c1],
1282 PrimitiveArray::empty::<i32>(Nullability::NonNullable)
1283 .dtype()
1284 .clone(),
1285 )?;
1286
1287 let indices =
1289 PrimitiveArray::from_option_iter([Some(5u64), None, Some(0), Some(3), None, Some(2)]);
1290 let result = arr.take(indices.into_array())?;
1291
1292 assert_arrays_eq!(
1293 result,
1294 PrimitiveArray::from_option_iter([
1295 Some(60i32),
1296 None,
1297 Some(10),
1298 Some(40),
1299 None,
1300 Some(30)
1301 ]),
1302 &mut ctx
1303 );
1304 Ok(())
1305 }
1306
1307 #[test]
1308 fn test_take_sorted_indices() -> VortexResult<()> {
1309 let mut ctx = array_session().create_execution_ctx();
1310 let arr = chunked_i32()?;
1311
1312 let indices = buffer![1u64, 4, 5, 9, 10, 14].into_array();
1314 let result = arr.take(indices)?;
1315
1316 assert_arrays_eq!(
1317 result,
1318 PrimitiveArray::from_iter([1i32, 4, 5, 9, 10, 14]),
1319 &mut ctx
1320 );
1321 Ok(())
1322 }
1323
1324 #[test]
1325 fn test_take_out_of_bounds() -> VortexResult<()> {
1326 let mut ctx = array_session().create_execution_ctx();
1327 let arr = chunked_i32()?;
1328
1329 for indices in [buffer![0u64, 100], buffer![100u64, 0]] {
1331 let result = arr
1332 .take(indices.into_array())
1333 .and_then(|taken| taken.execute::<Canonical>(&mut ctx));
1334 assert!(result.is_err(), "expected out-of-bounds error");
1335 }
1336 Ok(())
1337 }
1338
1339 #[test]
1340 fn test_take_all_null_indices() -> VortexResult<()> {
1341 let mut ctx = array_session().create_execution_ctx();
1342 let arr = chunked_i32()?;
1343
1344 let indices = PrimitiveArray::from_option_iter([None::<u64>, None]).into_array();
1345 let result = arr.take(indices)?;
1346
1347 assert_arrays_eq!(
1348 result,
1349 PrimitiveArray::from_option_iter([None::<i32>, None]),
1350 &mut ctx
1351 );
1352 Ok(())
1353 }
1354
1355 #[test]
1356 fn test_take_chunked_conformance() {
1357 let a = buffer![1i32, 2, 3].into_array();
1358 let b = buffer![4i32, 5].into_array();
1359 let arr = ChunkedArray::try_new(
1360 vec![a, b],
1361 PrimitiveArray::empty::<i32>(Nullability::NonNullable)
1362 .dtype()
1363 .clone(),
1364 )
1365 .unwrap();
1366 test_take_conformance(
1367 &arr.into_array(),
1368 &mut array_session().create_execution_ctx(),
1369 );
1370
1371 let a = PrimitiveArray::from_option_iter([Some(1i32), None, Some(3)]);
1373 let b = PrimitiveArray::from_option_iter([Some(4i32), Some(5)]);
1374 let dtype = a.dtype().clone();
1375 let arr = ChunkedArray::try_new(vec![a.into_array(), b.into_array()], dtype).unwrap();
1376 test_take_conformance(
1377 &arr.into_array(),
1378 &mut array_session().create_execution_ctx(),
1379 );
1380
1381 let chunk = buffer![10i32, 20, 30, 40, 50].into_array();
1383 let arr = ChunkedArray::try_new(
1384 vec![chunk.clone(), chunk.clone(), chunk.clone()],
1385 chunk.dtype().clone(),
1386 )
1387 .unwrap();
1388 test_take_conformance(
1389 &arr.into_array(),
1390 &mut array_session().create_execution_ctx(),
1391 );
1392 }
1393}