1use arrow::array::{
21 Array, ArrayRef, Capacities, GenericListArray, GenericListViewArray, Int64Array,
22 MutableArrayData, NullArray, OffsetSizeTrait,
23};
24use arrow::buffer::{NullBuffer, OffsetBuffer, ScalarBuffer};
25use arrow::datatypes::DataType;
26use arrow::datatypes::DataType::{
27 FixedSizeList, LargeList, LargeListView, List, ListView, Null,
28};
29use datafusion_common::cast::as_large_list_array;
30use datafusion_common::cast::as_list_array;
31use datafusion_common::cast::{
32 as_int64_array, as_large_list_view_array, as_list_view_array,
33};
34use datafusion_common::internal_err;
35use datafusion_common::utils::ListCoercion;
36use datafusion_common::{
37 Result, exec_datafusion_err, exec_err, internal_datafusion_err, plan_err,
38 utils::take_function_args,
39};
40use datafusion_expr::{
41 ArrayFunctionArgument, ArrayFunctionSignature, Expr, ScalarFunctionArgs,
42 TypeSignature,
43};
44use datafusion_expr::{
45 ColumnarValue, Documentation, ScalarUDFImpl, Signature, Volatility,
46};
47use datafusion_macros::user_doc;
48use std::sync::Arc;
49
50use crate::utils::{list_inner_field, make_scalar_function};
51
52make_udf_expr_and_func!(
54 ArrayElement,
55 array_element,
56 array element,
57 "extracts the element with the index n from the array.",
58 array_element_udf
59);
60
61create_func!(ArraySlice, array_slice_udf);
62
63make_udf_expr_and_func!(
64 ArrayPopFront,
65 array_pop_front,
66 array,
67 "returns the array without the first element.",
68 array_pop_front_udf
69);
70
71make_udf_expr_and_func!(
72 ArrayPopBack,
73 array_pop_back,
74 array,
75 "returns the array without the last element.",
76 array_pop_back_udf
77);
78
79make_udf_expr_and_func!(
80 ArrayAnyValue,
81 array_any_value,
82 array,
83 "returns the first non-null element in the array.",
84 array_any_value_udf
85);
86
87#[user_doc(
88 doc_section(label = "Array Functions"),
89 description = "Extracts the element with the index n from the array.",
90 syntax_example = "array_element(array, index)",
91 sql_example = r#"```sql
92> select array_element([1, 2, 3, 4], 3);
93+-----------------------------------------+
94| array_element(List([1,2,3,4]),Int64(3)) |
95+-----------------------------------------+
96| 3 |
97+-----------------------------------------+
98```"#,
99 argument(
100 name = "array",
101 description = "Array expression. Can be a constant, column, or function, and any combination of array operators."
102 ),
103 argument(
104 name = "index",
105 description = "Index to extract the element from the array."
106 )
107)]
108#[derive(Debug, PartialEq, Eq, Hash)]
109pub struct ArrayElement {
110 signature: Signature,
111 aliases: Vec<String>,
112}
113
114impl Default for ArrayElement {
115 fn default() -> Self {
116 Self::new()
117 }
118}
119
120impl ArrayElement {
121 pub fn new() -> Self {
122 Self {
123 signature: Signature::array_and_index(Volatility::Immutable),
124 aliases: vec![
125 String::from("array_extract"),
126 String::from("list_element"),
127 String::from("list_extract"),
128 ],
129 }
130 }
131}
132
133impl ScalarUDFImpl for ArrayElement {
134 fn name(&self) -> &str {
135 "array_element"
136 }
137
138 fn display_name(&self, args: &[Expr]) -> Result<String> {
139 let args_name = args.iter().map(ToString::to_string).collect::<Vec<_>>();
140 if args_name.len() != 2 {
141 return exec_err!("expect 2 args, got {}", args_name.len());
142 }
143
144 Ok(format!("{}[{}]", args_name[0], args_name[1]))
145 }
146
147 fn schema_name(&self, args: &[Expr]) -> Result<String> {
148 let args_name = args
149 .iter()
150 .map(|e| e.schema_name().to_string())
151 .collect::<Vec<_>>();
152 if args_name.len() != 2 {
153 return exec_err!("expect 2 args, got {}", args_name.len());
154 }
155
156 Ok(format!("{}[{}]", args_name[0], args_name[1]))
157 }
158
159 fn signature(&self) -> &Signature {
160 &self.signature
161 }
162
163 fn return_type(&self, arg_types: &[DataType]) -> Result<DataType> {
164 match &arg_types[0] {
165 Null => Ok(Null),
166 List(field) | LargeList(field) => Ok(field.data_type().clone()),
167 arg_type => plan_err!("{} does not support type {arg_type}", self.name()),
168 }
169 }
170
171 fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result<ColumnarValue> {
172 make_scalar_function(array_element_inner)(&args.args)
173 }
174
175 fn aliases(&self) -> &[String] {
176 &self.aliases
177 }
178
179 fn documentation(&self) -> Option<&Documentation> {
180 self.doc()
181 }
182}
183
184fn array_element_inner(args: &[ArrayRef]) -> Result<ArrayRef> {
192 let [array, indexes] = take_function_args("array_element", args)?;
193
194 match &array.data_type() {
195 Null => Ok(Arc::new(NullArray::new(array.len()))),
196 List(_) => {
197 let array = as_list_array(&array)?;
198 let indexes = as_int64_array(&indexes)?;
199 general_array_element::<i32>(array, indexes)
200 }
201 LargeList(_) => {
202 let array = as_large_list_array(&array)?;
203 let indexes = as_int64_array(&indexes)?;
204 general_array_element::<i64>(array, indexes)
205 }
206 arg_type => {
207 exec_err!("array_element does not support type {arg_type}")
208 }
209 }
210}
211
212fn general_array_element<O: OffsetSizeTrait>(
213 array: &GenericListArray<O>,
214 indexes: &Int64Array,
215) -> Result<ArrayRef>
216where
217 i64: TryInto<O>,
218{
219 let values = array.values();
220 if values.data_type().is_null() {
221 return Ok(Arc::new(NullArray::new(array.len())));
222 }
223
224 let original_data = values.to_data();
225 let capacity = Capacities::Array(original_data.len());
226
227 let mut mutable =
229 MutableArrayData::with_capacities(vec![&original_data], true, capacity);
230
231 fn adjusted_array_index<O: OffsetSizeTrait>(index: i64, len: O) -> Result<Option<O>>
232 where
233 i64: TryInto<O>,
234 {
235 let index: O = index.try_into().map_err(|_| {
236 exec_datafusion_err!("array_element got invalid index: {index}")
237 })?;
238 let adjusted_zero_index = if index < O::usize_as(0) {
240 index + len
241 } else {
242 index - O::usize_as(1)
243 };
244
245 if O::usize_as(0) <= adjusted_zero_index && adjusted_zero_index < len {
246 Ok(Some(adjusted_zero_index))
247 } else {
248 Ok(None)
250 }
251 }
252
253 for (row_index, offset_window) in array.offsets().windows(2).enumerate() {
254 let start = offset_window[0];
255 let end = offset_window[1];
256 let len = end - start;
257
258 if array.is_null(row_index) || indexes.is_null(row_index) {
260 mutable.try_extend_nulls(1)?;
261 continue;
262 }
263
264 let index = adjusted_array_index::<O>(indexes.value(row_index), len)?;
265
266 if let Some(index) = index {
267 let start = start.as_usize() + index.as_usize();
268 mutable.try_extend(0, start, start + 1_usize)?;
269 } else {
270 mutable.try_extend_nulls(1)?;
272 }
273 }
274
275 let data = mutable.freeze();
276 Ok(arrow::array::make_array(data))
277}
278
279#[doc = "returns a slice of the array."]
280pub fn array_slice(array: Expr, begin: Expr, end: Expr, stride: Option<Expr>) -> Expr {
281 let args = match stride {
282 Some(stride) => vec![array, begin, end, stride],
283 None => vec![array, begin, end],
284 };
285 array_slice_udf().call(args)
286}
287
288#[user_doc(
289 doc_section(label = "Array Functions"),
290 description = "Returns a slice of the array based on 1-indexed start and end positions.",
291 syntax_example = "array_slice(array, begin, end[, stride])",
292 sql_example = r#"```sql
293> select array_slice([1, 2, 3, 4, 5, 6, 7, 8], 3, 6);
294+--------------------------------------------------------+
295| array_slice(List([1,2,3,4,5,6,7,8]),Int64(3),Int64(6)) |
296+--------------------------------------------------------+
297| [3, 4, 5, 6] |
298+--------------------------------------------------------+
299```"#,
300 argument(
301 name = "array",
302 description = "Array expression. Can be a constant, column, or function, and any combination of array operators."
303 ),
304 argument(
305 name = "begin",
306 description = "Index of the first element. If negative, it counts backward from the end of the array."
307 ),
308 argument(
309 name = "end",
310 description = "Index of the last element. If negative, it counts backward from the end of the array."
311 ),
312 argument(
313 name = "stride",
314 description = "Stride of the array slice. The default is 1."
315 )
316)]
317#[derive(Debug, PartialEq, Eq, Hash)]
318pub(super) struct ArraySlice {
319 signature: Signature,
320 aliases: Vec<String>,
321}
322
323impl ArraySlice {
324 pub fn new() -> Self {
325 Self {
326 signature: Signature::one_of(
327 vec![
328 TypeSignature::ArraySignature(ArrayFunctionSignature::Array {
329 arguments: vec![
330 ArrayFunctionArgument::Array,
331 ArrayFunctionArgument::Index,
332 ArrayFunctionArgument::Index,
333 ],
334 array_coercion: Some(ListCoercion::FixedSizedListToList),
335 }),
336 TypeSignature::ArraySignature(ArrayFunctionSignature::Array {
337 arguments: vec![
338 ArrayFunctionArgument::Array,
339 ArrayFunctionArgument::Index,
340 ArrayFunctionArgument::Index,
341 ArrayFunctionArgument::Index,
342 ],
343 array_coercion: Some(ListCoercion::FixedSizedListToList),
344 }),
345 ],
346 Volatility::Immutable,
347 ),
348 aliases: vec![String::from("list_slice")],
349 }
350 }
351}
352
353impl ScalarUDFImpl for ArraySlice {
354 fn display_name(&self, args: &[Expr]) -> Result<String> {
355 let args_name = args.iter().map(ToString::to_string).collect::<Vec<_>>();
356 if let Some((arr, indexes)) = args_name.split_first() {
357 Ok(format!("{arr}[{}]", indexes.join(":")))
358 } else {
359 exec_err!("no argument")
360 }
361 }
362
363 fn schema_name(&self, args: &[Expr]) -> Result<String> {
364 let args_name = args
365 .iter()
366 .map(|e| e.schema_name().to_string())
367 .collect::<Vec<_>>();
368 if let Some((arr, indexes)) = args_name.split_first() {
369 Ok(format!("{arr}[{}]", indexes.join(":")))
370 } else {
371 exec_err!("no argument")
372 }
373 }
374
375 fn name(&self) -> &str {
376 "array_slice"
377 }
378
379 fn signature(&self) -> &Signature {
380 &self.signature
381 }
382
383 fn return_type(&self, arg_types: &[DataType]) -> Result<DataType> {
384 Ok(arg_types[0].clone())
385 }
386
387 fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result<ColumnarValue> {
388 make_scalar_function(array_slice_inner)(&args.args)
389 }
390
391 fn aliases(&self) -> &[String] {
392 &self.aliases
393 }
394
395 fn documentation(&self) -> Option<&Documentation> {
396 self.doc()
397 }
398}
399
400fn array_slice_inner(args: &[ArrayRef]) -> Result<ArrayRef> {
417 let args_len = args.len();
418 if args_len != 3 && args_len != 4 {
419 return exec_err!("array_slice needs three or four arguments");
420 }
421
422 let stride = if args_len == 4 {
423 Some(as_int64_array(&args[3])?)
424 } else {
425 None
426 };
427
428 let from_array = as_int64_array(&args[1])?;
429 let to_array = as_int64_array(&args[2])?;
430
431 let array_data_type = args[0].data_type();
432 match array_data_type {
433 List(_) => {
434 let array = as_list_array(&args[0])?;
435 general_array_slice::<i32>(array, from_array, to_array, stride)
436 }
437 LargeList(_) => {
438 let array = as_large_list_array(&args[0])?;
439 general_array_slice::<i64>(array, from_array, to_array, stride)
440 }
441 ListView(_) => {
442 let array = as_list_view_array(&args[0])?;
443 general_list_view_array_slice::<i32>(array, from_array, to_array, stride)
444 }
445 LargeListView(_) => {
446 let array = as_large_list_view_array(&args[0])?;
447 general_list_view_array_slice::<i64>(array, from_array, to_array, stride)
448 }
449 _ => exec_err!("array_slice does not support type: {}", array_data_type),
450 }
451}
452
453fn adjusted_from_index<O: OffsetSizeTrait>(index: i64, len: O) -> Result<Option<O>>
454where
455 i64: TryInto<O>,
456{
457 let adjusted_zero_index = if index < 0 {
459 if let Ok(index) = index.try_into() {
460 if index < (O::zero() - O::one()) * len {
467 O::zero()
468 } else {
469 index + len
470 }
471 } else {
472 return exec_err!("array_slice got invalid index: {}", index);
473 }
474 } else {
475 if let Ok(index) = index.try_into() {
477 std::cmp::max(index - O::usize_as(1), O::usize_as(0))
478 } else {
479 return exec_err!("array_slice got invalid index: {}", index);
480 }
481 };
482
483 if O::usize_as(0) <= adjusted_zero_index && adjusted_zero_index < len {
484 Ok(Some(adjusted_zero_index))
485 } else {
486 Ok(None)
488 }
489}
490
491fn adjusted_to_index<O: OffsetSizeTrait>(index: i64, len: O) -> Result<Option<O>>
492where
493 i64: TryInto<O>,
494{
495 let adjusted_zero_index = if index < 0 {
497 if let Ok(index) = index.try_into() {
499 index + len
500 } else {
501 return exec_err!("array_slice got invalid index: {}", index);
502 }
503 } else {
504 if let Ok(index) = index.try_into() {
506 std::cmp::min(index - O::usize_as(1), len - O::usize_as(1))
507 } else {
508 return exec_err!("array_slice got invalid index: {}", index);
509 }
510 };
511
512 if O::usize_as(0) <= adjusted_zero_index && adjusted_zero_index < len {
513 Ok(Some(adjusted_zero_index))
514 } else {
515 Ok(None)
517 }
518}
519
520enum SlicePlan<O: OffsetSizeTrait> {
524 Empty,
526 Contiguous { start: O, len: O },
529 Indices(Vec<O>),
532}
533
534fn compute_slice_plan<O: OffsetSizeTrait>(
536 len: O,
537 from_raw: i64,
538 to_raw: i64,
539 stride_raw: Option<i64>,
540) -> Result<SlicePlan<O>>
541where
542 i64: TryInto<O>,
543{
544 if len == O::usize_as(0) {
545 return Ok(SlicePlan::Empty);
546 }
547
548 let from_index = adjusted_from_index::<O>(from_raw, len)?;
549 let to_index = adjusted_to_index::<O>(to_raw, len)?;
550
551 let (Some(from), Some(to)) = (from_index, to_index) else {
552 return Ok(SlicePlan::Empty);
553 };
554
555 let stride_value = stride_raw.unwrap_or(1);
556 if stride_value == 0 {
557 return exec_err!(
558 "array_slice got invalid stride: {:?}, it cannot be 0",
559 stride_value
560 );
561 }
562
563 if (from < to && stride_value.is_negative())
564 || (from > to && stride_value.is_positive())
565 {
566 return Ok(SlicePlan::Empty);
567 }
568
569 let stride: O = stride_value.try_into().map_err(|_| {
570 internal_datafusion_err!("array_slice got invalid stride: {}", stride_value)
571 })?;
572
573 if from <= to && stride_value.is_positive() {
574 if stride_value == 1 {
575 let len = to - from + O::usize_as(1);
576 Ok(SlicePlan::Contiguous { start: from, len })
577 } else {
578 let mut indices = Vec::new();
579 let mut index = from;
580 while index <= to {
581 indices.push(index);
582 index += stride;
583 }
584 Ok(SlicePlan::Indices(indices))
585 }
586 } else {
587 let mut indices = Vec::new();
588 let mut index = from;
589 while index >= to {
590 indices.push(index);
591 index += stride;
592 }
593 Ok(SlicePlan::Indices(indices))
594 }
595}
596
597fn combine_input_nulls(
599 array: &dyn Array,
600 from_array: &Int64Array,
601 to_array: &Int64Array,
602 stride: Option<&Int64Array>,
603) -> Option<NullBuffer> {
604 NullBuffer::union_many([
605 array.nulls(),
606 from_array.nulls(),
607 to_array.nulls(),
608 stride.and_then(|s| s.nulls()),
609 ])
610}
611
612fn general_array_slice<O: OffsetSizeTrait>(
613 array: &GenericListArray<O>,
614 from_array: &Int64Array,
615 to_array: &Int64Array,
616 stride: Option<&Int64Array>,
617) -> Result<ArrayRef>
618where
619 i64: TryInto<O>,
620{
621 let values = array.values();
622 let original_data = values.to_data();
623 let capacity = Capacities::Array(original_data.len());
624 let field = list_inner_field("general_array_slice", array.data_type())?;
628
629 let mut mutable =
633 MutableArrayData::with_capacities(vec![&original_data], false, capacity);
634
635 let mut offsets = vec![O::usize_as(0)];
639
640 let nulls = combine_input_nulls(array, from_array, to_array, stride);
641
642 for (row_index, offset_window) in array.offsets().windows(2).enumerate() {
643 let start = offset_window[0];
644 let end = offset_window[1];
645 let len = end - start;
646
647 if nulls.as_ref().is_some_and(|n| n.is_null(row_index)) {
651 offsets.push(offsets[row_index]);
652 continue;
653 }
654
655 if len == O::usize_as(0) {
657 offsets.push(offsets[row_index]);
658 continue;
659 }
660
661 let slice_plan = compute_slice_plan::<O>(
662 len,
663 from_array.value(row_index),
664 to_array.value(row_index),
665 stride.map(|s| s.value(row_index)),
666 )?;
667
668 match slice_plan {
669 SlicePlan::Empty => offsets.push(offsets[row_index]),
670 SlicePlan::Contiguous {
671 start: rel_start,
672 len: slice_len,
673 } => {
674 let start_index = (start + rel_start).to_usize().unwrap();
675 let end_index = (start + rel_start + slice_len).to_usize().unwrap();
676 mutable.try_extend(0, start_index, end_index)?;
677 offsets.push(offsets[row_index] + slice_len);
678 }
679 SlicePlan::Indices(indices) => {
680 let count = indices.len();
681 for rel_index in indices {
682 let absolute_index = (start + rel_index).to_usize().unwrap();
683 mutable.try_extend(0, absolute_index, absolute_index + 1)?;
684 }
685 offsets.push(offsets[row_index] + O::usize_as(count));
686 }
687 }
688 }
689
690 let data = mutable.freeze();
691
692 Ok(Arc::new(GenericListArray::<O>::try_new(
693 field,
694 OffsetBuffer::<O>::new(offsets.into()),
695 arrow::array::make_array(data),
696 nulls,
697 )?))
698}
699
700fn general_list_view_array_slice<O: OffsetSizeTrait>(
701 array: &GenericListViewArray<O>,
702 from_array: &Int64Array,
703 to_array: &Int64Array,
704 stride: Option<&Int64Array>,
705) -> Result<ArrayRef>
706where
707 i64: TryInto<O>,
708{
709 let values = array.values();
710 let original_data = values.to_data();
711 let capacity = Capacities::Array(original_data.len());
712 let field = match array.data_type() {
713 ListView(field) | LargeListView(field) => Arc::clone(field),
714 other => {
715 return internal_err!(
716 "general_list_view_array_slice got unexpected data type: {other}"
717 );
718 }
719 };
720
721 let mut mutable =
723 MutableArrayData::with_capacities(vec![&original_data], false, capacity);
724
725 let mut offsets = Vec::with_capacity(array.len());
728 let mut sizes = Vec::with_capacity(array.len());
729 let mut current_offset = O::usize_as(0);
730
731 let nulls = combine_input_nulls(array, from_array, to_array, stride);
732
733 for row_index in 0..array.len() {
734 if nulls.as_ref().is_some_and(|n| n.is_null(row_index)) {
735 offsets.push(current_offset);
736 sizes.push(O::usize_as(0));
737 continue;
738 }
739
740 let len = array.value_size(row_index);
741
742 if len == O::usize_as(0) {
744 offsets.push(current_offset);
745 sizes.push(O::usize_as(0));
746 continue;
747 }
748
749 let slice_plan = compute_slice_plan::<O>(
750 len,
751 from_array.value(row_index),
752 to_array.value(row_index),
753 stride.map(|s| s.value(row_index)),
754 )?;
755
756 let start = array.value_offset(row_index);
757 match slice_plan {
758 SlicePlan::Empty => {
759 offsets.push(current_offset);
760 sizes.push(O::usize_as(0));
761 }
762 SlicePlan::Contiguous {
763 start: rel_start,
764 len: slice_len,
765 } => {
766 let start_index = (start + rel_start).to_usize().unwrap();
767 let end_index = (start + rel_start + slice_len).to_usize().unwrap();
768 mutable.try_extend(0, start_index, end_index)?;
769 offsets.push(current_offset);
770 sizes.push(slice_len);
771 current_offset += slice_len;
772 }
773 SlicePlan::Indices(indices) => {
774 let count = indices.len();
775 for rel_index in indices {
776 let absolute_index = (start + rel_index).to_usize().unwrap();
777 mutable.try_extend(0, absolute_index, absolute_index + 1)?;
778 }
779 let length = O::usize_as(count);
780 offsets.push(current_offset);
781 sizes.push(length);
782 current_offset += length;
783 }
784 }
785 }
786
787 let data = mutable.freeze();
788
789 Ok(Arc::new(GenericListViewArray::<O>::try_new(
790 field,
791 ScalarBuffer::from(offsets),
792 ScalarBuffer::from(sizes),
793 arrow::array::make_array(data),
794 nulls,
795 )?))
796}
797
798#[user_doc(
799 doc_section(label = "Array Functions"),
800 description = "Returns the array without the first element.",
801 syntax_example = "array_pop_front(array)",
802 sql_example = r#"```sql
803> select array_pop_front([1, 2, 3]);
804+-------------------------------+
805| array_pop_front(List([1,2,3])) |
806+-------------------------------+
807| [2, 3] |
808+-------------------------------+
809```"#,
810 argument(
811 name = "array",
812 description = "Array expression. Can be a constant, column, or function, and any combination of array operators."
813 )
814)]
815#[derive(Debug, PartialEq, Eq, Hash)]
816pub(super) struct ArrayPopFront {
817 signature: Signature,
818 aliases: Vec<String>,
819}
820
821impl ArrayPopFront {
822 pub fn new() -> Self {
823 Self {
824 signature: Signature::array(Volatility::Immutable),
825 aliases: vec![String::from("list_pop_front")],
826 }
827 }
828}
829
830impl ScalarUDFImpl for ArrayPopFront {
831 fn name(&self) -> &str {
832 "array_pop_front"
833 }
834
835 fn signature(&self) -> &Signature {
836 &self.signature
837 }
838
839 fn return_type(&self, arg_types: &[DataType]) -> Result<DataType> {
840 Ok(arg_types[0].clone())
841 }
842
843 fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result<ColumnarValue> {
844 make_scalar_function(array_pop_front_inner)(&args.args)
845 }
846
847 fn aliases(&self) -> &[String] {
848 &self.aliases
849 }
850
851 fn documentation(&self) -> Option<&Documentation> {
852 self.doc()
853 }
854}
855
856fn array_pop_front_inner(args: &[ArrayRef]) -> Result<ArrayRef> {
858 let array_data_type = args[0].data_type();
859 match array_data_type {
860 List(_) => {
861 let array = as_list_array(&args[0])?;
862 general_pop_front_list::<i32>(array)
863 }
864 LargeList(_) => {
865 let array = as_large_list_array(&args[0])?;
866 general_pop_front_list::<i64>(array)
867 }
868 _ => exec_err!("array_pop_front does not support type: {}", array_data_type),
869 }
870}
871
872fn general_pop_front_list<O: OffsetSizeTrait>(
873 array: &GenericListArray<O>,
874) -> Result<ArrayRef>
875where
876 i64: TryInto<O>,
877{
878 let from_array = Int64Array::from(vec![2; array.len()]);
879 let to_array = Int64Array::from(
880 array
881 .iter()
882 .map(|arr| arr.map_or(0, |arr| arr.len() as i64))
883 .collect::<Vec<i64>>(),
884 );
885 general_array_slice::<O>(array, &from_array, &to_array, None)
886}
887
888#[user_doc(
889 doc_section(label = "Array Functions"),
890 description = "Returns the array without the last element.",
891 syntax_example = "array_pop_back(array)",
892 sql_example = r#"```sql
893> select array_pop_back([1, 2, 3]);
894+-------------------------------+
895| array_pop_back(List([1,2,3])) |
896+-------------------------------+
897| [1, 2] |
898+-------------------------------+
899```"#,
900 argument(
901 name = "array",
902 description = "Array expression. Can be a constant, column, or function, and any combination of array operators."
903 )
904)]
905#[derive(Debug, PartialEq, Eq, Hash)]
906pub(super) struct ArrayPopBack {
907 signature: Signature,
908 aliases: Vec<String>,
909}
910
911impl ArrayPopBack {
912 pub fn new() -> Self {
913 Self {
914 signature: Signature::array(Volatility::Immutable),
915 aliases: vec![String::from("list_pop_back")],
916 }
917 }
918}
919
920impl ScalarUDFImpl for ArrayPopBack {
921 fn name(&self) -> &str {
922 "array_pop_back"
923 }
924
925 fn signature(&self) -> &Signature {
926 &self.signature
927 }
928
929 fn return_type(&self, arg_types: &[DataType]) -> Result<DataType> {
930 Ok(arg_types[0].clone())
931 }
932
933 fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result<ColumnarValue> {
934 make_scalar_function(array_pop_back_inner)(&args.args)
935 }
936
937 fn aliases(&self) -> &[String] {
938 &self.aliases
939 }
940
941 fn documentation(&self) -> Option<&Documentation> {
942 self.doc()
943 }
944}
945
946fn array_pop_back_inner(args: &[ArrayRef]) -> Result<ArrayRef> {
948 let [array] = take_function_args("array_pop_back", args)?;
949
950 match array.data_type() {
951 List(_) => {
952 let array = as_list_array(&array)?;
953 general_pop_back_list::<i32>(array)
954 }
955 LargeList(_) => {
956 let array = as_large_list_array(&array)?;
957 general_pop_back_list::<i64>(array)
958 }
959 _ => exec_err!(
960 "array_pop_back does not support type: {}",
961 array.data_type()
962 ),
963 }
964}
965
966fn general_pop_back_list<O: OffsetSizeTrait>(
967 array: &GenericListArray<O>,
968) -> Result<ArrayRef>
969where
970 i64: TryInto<O>,
971{
972 let from_array = Int64Array::from(vec![1; array.len()]);
973 let to_array = Int64Array::from(
974 array
975 .iter()
976 .map(|arr| arr.map_or(0, |arr| arr.len() as i64 - 1))
977 .collect::<Vec<i64>>(),
978 );
979 general_array_slice::<O>(array, &from_array, &to_array, None)
980}
981
982#[user_doc(
983 doc_section(label = "Array Functions"),
984 description = "Returns the first non-null element in the array. Returns NULL if the array is empty or NULL.",
985 syntax_example = "array_any_value(array)",
986 sql_example = r#"```sql
987> select array_any_value([NULL, 1, 2, 3]);
988+-------------------------------+
989| array_any_value(List([NULL,1,2,3])) |
990+-------------------------------------+
991| 1 |
992+-------------------------------------+
993```"#,
994 argument(
995 name = "array",
996 description = "Array expression. Can be a constant, column, or function, and any combination of array operators."
997 )
998)]
999#[derive(Debug, PartialEq, Eq, Hash)]
1000pub(super) struct ArrayAnyValue {
1001 signature: Signature,
1002 aliases: Vec<String>,
1003}
1004
1005impl ArrayAnyValue {
1006 pub fn new() -> Self {
1007 Self {
1008 signature: Signature::array(Volatility::Immutable),
1009 aliases: vec![String::from("list_any_value")],
1010 }
1011 }
1012}
1013
1014impl ScalarUDFImpl for ArrayAnyValue {
1015 fn name(&self) -> &str {
1016 "array_any_value"
1017 }
1018 fn signature(&self) -> &Signature {
1019 &self.signature
1020 }
1021 fn return_type(&self, arg_types: &[DataType]) -> Result<DataType> {
1022 match &arg_types[0] {
1023 List(field) | LargeList(field) | FixedSizeList(field, _) => {
1024 Ok(field.data_type().clone())
1025 }
1026 _ => plan_err!(
1027 "array_any_value can only accept List, LargeList or FixedSizeList as the argument"
1028 ),
1029 }
1030 }
1031
1032 fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result<ColumnarValue> {
1033 make_scalar_function(array_any_value_inner)(&args.args)
1034 }
1035
1036 fn aliases(&self) -> &[String] {
1037 &self.aliases
1038 }
1039
1040 fn documentation(&self) -> Option<&Documentation> {
1041 self.doc()
1042 }
1043}
1044
1045fn array_any_value_inner(args: &[ArrayRef]) -> Result<ArrayRef> {
1046 let [array] = take_function_args("array_any_value", args)?;
1047
1048 match &array.data_type() {
1049 List(_) => {
1050 let array = as_list_array(&array)?;
1051 general_array_any_value::<i32>(array)
1052 }
1053 LargeList(_) => {
1054 let array = as_large_list_array(&array)?;
1055 general_array_any_value::<i64>(array)
1056 }
1057 data_type => exec_err!("array_any_value does not support type: {data_type}"),
1058 }
1059}
1060
1061fn general_array_any_value<O: OffsetSizeTrait>(
1062 array: &GenericListArray<O>,
1063) -> Result<ArrayRef>
1064where
1065 i64: TryInto<O>,
1066{
1067 let values = array.values();
1068 let original_data = values.to_data();
1069 let capacity = Capacities::Array(array.len());
1070
1071 let mut mutable =
1072 MutableArrayData::with_capacities(vec![&original_data], true, capacity);
1073
1074 for (row_index, offset_window) in array.offsets().windows(2).enumerate() {
1075 let start = offset_window[0];
1076 let end = offset_window[1];
1077
1078 if array.is_null(row_index) {
1080 mutable.try_extend_nulls(1)?;
1081 continue;
1082 }
1083
1084 if start == end {
1087 mutable.try_extend_nulls(1)?;
1088 continue;
1089 }
1090
1091 let row_value = array.value(row_index);
1092 match row_value.nulls() {
1093 Some(row_nulls_buffer) => {
1094 if let Some(first_non_null_index) =
1096 row_nulls_buffer.valid_indices().next()
1097 {
1098 let index = start.as_usize() + first_non_null_index;
1099 mutable.try_extend(0, index, index + 1)?;
1100 } else {
1101 mutable.try_extend_nulls(1)?;
1103 }
1104 }
1105 None => {
1106 let index = start.as_usize();
1108 mutable.try_extend(0, index, index + 1)?;
1109 }
1110 }
1111 }
1112
1113 let data = mutable.freeze();
1114 Ok(arrow::array::make_array(data))
1115}
1116
1117#[cfg(test)]
1118mod tests {
1119 use super::{
1120 array_element_udf, general_array_any_value, general_array_element,
1121 general_list_view_array_slice,
1122 };
1123 use arrow::array::{
1124 Array, ArrayRef, GenericListViewArray, Int32Array, Int64Array, ListViewArray,
1125 cast::AsArray,
1126 };
1127 use arrow::array::{ListArray, RecordBatch};
1128 use arrow::buffer::{NullBuffer, OffsetBuffer, ScalarBuffer};
1129 use arrow::datatypes::{DataType, Field, Int32Type};
1130 use datafusion_common::{Column, DFSchema, Result, assert_batches_eq};
1131 use datafusion_expr::expr::ScalarFunction;
1132 use datafusion_expr::{Expr, ExprSchemable};
1133 use std::collections::HashMap;
1134 use std::sync::Arc;
1135
1136 fn list_view_values(array: &GenericListViewArray<i32>) -> Vec<Vec<i32>> {
1137 (0..array.len())
1138 .map(|i| {
1139 let child = array.value(i);
1140 let values = child.as_any().downcast_ref::<Int32Array>().unwrap();
1141 values.iter().map(|v| v.unwrap()).collect()
1142 })
1143 .collect()
1144 }
1145
1146 #[test]
1148 fn test_array_element_return_type_fixed_size_list() {
1149 let fixed_size_list_type = DataType::FixedSizeList(
1150 Field::new("some_arbitrary_test_field", DataType::Int32, false).into(),
1151 13,
1152 );
1153 let array_type = DataType::List(
1154 Field::new_list_field(fixed_size_list_type.clone(), true).into(),
1155 );
1156 let index_type = DataType::Int64;
1157
1158 let schema = DFSchema::from_unqualified_fields(
1159 vec![
1160 Field::new("my_array", array_type.clone(), false),
1161 Field::new("my_index", index_type.clone(), false),
1162 ]
1163 .into(),
1164 HashMap::default(),
1165 )
1166 .unwrap();
1167
1168 let udf = array_element_udf();
1169
1170 assert_eq!(
1172 udf.return_type(&[array_type.clone(), index_type.clone()])
1173 .unwrap(),
1174 fixed_size_list_type
1175 );
1176
1177 let udf_expr = Expr::ScalarFunction(ScalarFunction {
1179 func: array_element_udf(),
1180 args: vec![
1181 Expr::Column(Column::new_unqualified("my_array")),
1182 Expr::Column(Column::new_unqualified("my_index")),
1183 ],
1184 });
1185 assert_eq!(
1186 ExprSchemable::get_type(&udf_expr, &schema).unwrap(),
1187 fixed_size_list_type
1188 );
1189 }
1190
1191 #[test]
1192 fn test_array_element_null_handling() -> Result<()> {
1193 let values = Arc::new(Int32Array::from(vec![1, 2, 3, 4, 5]));
1194 let offsets = OffsetBuffer::new(ScalarBuffer::from(vec![0, 3, 4, 5]));
1195 let nulls = NullBuffer::from(vec![true, false, true]);
1196 let field = Arc::new(Field::new("item", DataType::Int32, true));
1197
1198 let list_array = ListArray::new(field, offsets, values, Some(nulls));
1199 let indexes = Int64Array::from(vec![1, 1, 1]);
1200
1201 let result = general_array_element(&list_array, &indexes)?;
1202
1203 let expected = [
1204 "+--------+",
1205 "| result |",
1206 "+--------+",
1207 "| 1 |",
1208 "| |",
1209 "| 5 |",
1210 "+--------+",
1211 ];
1212
1213 let batch = RecordBatch::try_from_iter([("result", result)])?;
1214
1215 assert_batches_eq!(expected, &[batch]);
1216
1217 Ok(())
1218 }
1219
1220 #[test]
1221 fn test_array_element_null_index_with_non_zero_buffer_returns_null() -> Result<()> {
1222 let list_array = ListArray::from_iter_primitive::<Int32Type, _, _>(vec![
1223 Some(vec![Some(1), Some(2), Some(3)]),
1224 Some(vec![Some(4)]),
1225 Some(vec![Some(5)]),
1226 ]);
1227 let indexes = Int64Array::new(
1228 ScalarBuffer::from(vec![1, 1, 1]),
1229 Some(NullBuffer::from(vec![true, false, true])),
1230 );
1231
1232 let result = general_array_element(&list_array, &indexes)?;
1233 let expected = Int32Array::from(vec![Some(1), None, Some(5)]);
1234
1235 assert_eq!(result.as_primitive::<Int32Type>(), &expected);
1236
1237 Ok(())
1238 }
1239
1240 #[test]
1241 fn test_array_any_null_handling() -> Result<()> {
1242 let values: ArrayRef = Arc::new(Int32Array::from(vec![1, 2, 3, 4, 5]));
1243 let offsets = OffsetBuffer::new(ScalarBuffer::from(vec![0, 3, 4, 5]));
1244 let nulls = NullBuffer::from(vec![true, false, true]);
1245 let field = Arc::new(Field::new("item", DataType::Int32, true));
1246
1247 let list_array = ListArray::new(field, offsets, values, Some(nulls));
1248
1249 let result = general_array_any_value(&list_array)?;
1250
1251 assert!(!result.is_null(0));
1252 assert!(result.is_null(1));
1253 assert!(!result.is_null(2));
1254
1255 Ok(())
1256 }
1257
1258 #[test]
1259 fn test_array_slice_list_view_basic() -> Result<()> {
1260 let values: ArrayRef = Arc::new(Int32Array::from(vec![1, 2, 3, 4, 5]));
1261 let offsets = ScalarBuffer::from(vec![0, 3]);
1262 let sizes = ScalarBuffer::from(vec![3, 2]);
1263 let field = Arc::new(Field::new("item", DataType::Int32, true));
1264 let array = ListViewArray::new(field, offsets, sizes, values, None);
1265
1266 let from = Int64Array::from(vec![2, 1]);
1267 let to = Int64Array::from(vec![3, 2]);
1268
1269 let result = general_list_view_array_slice::<i32>(
1270 &array,
1271 &from,
1272 &to,
1273 None::<&Int64Array>,
1274 )?;
1275 let result = result.as_ref().as_list_view::<i32>();
1276
1277 assert_eq!(list_view_values(result), vec![vec![2, 3], vec![4, 5]]);
1278 Ok(())
1279 }
1280
1281 #[test]
1282 fn test_array_slice_list_view_non_monotonic_offsets() -> Result<()> {
1283 let values: ArrayRef = Arc::new(Int32Array::from(vec![1, 2, 3, 4, 5]));
1285 let offsets = ScalarBuffer::from(vec![3, 0]);
1286 let sizes = ScalarBuffer::from(vec![2, 3]);
1287 let field = Arc::new(Field::new("item", DataType::Int32, true));
1288 let array = ListViewArray::new(field, offsets, sizes, values, None);
1289
1290 let from = Int64Array::from(vec![1, 1]);
1291 let to = Int64Array::from(vec![2, 2]);
1292
1293 let result = general_list_view_array_slice::<i32>(
1294 &array,
1295 &from,
1296 &to,
1297 None::<&Int64Array>,
1298 )?;
1299 let result = result.as_ref().as_list_view::<i32>();
1300
1301 assert_eq!(list_view_values(result), vec![vec![4, 5], vec![1, 2]]);
1302 Ok(())
1303 }
1304
1305 #[test]
1306 fn test_array_slice_list_view_negative_stride() -> Result<()> {
1307 let values: ArrayRef = Arc::new(Int32Array::from(vec![1, 2, 3, 4, 5]));
1308 let offsets = ScalarBuffer::from(vec![0, 3]);
1309 let sizes = ScalarBuffer::from(vec![3, 2]);
1310 let field = Arc::new(Field::new("item", DataType::Int32, true));
1311 let array = ListViewArray::new(field, offsets, sizes, values, None);
1312
1313 let from = Int64Array::from(vec![3, 2]);
1314 let to = Int64Array::from(vec![1, 1]);
1315 let stride = Int64Array::from(vec![-1, -1]);
1316
1317 let result =
1318 general_list_view_array_slice::<i32>(&array, &from, &to, Some(&stride))?;
1319 let result = result.as_ref().as_list_view::<i32>();
1320
1321 assert_eq!(list_view_values(result), vec![vec![3, 2, 1], vec![5, 4]]);
1322 Ok(())
1323 }
1324
1325 #[test]
1326 fn test_array_slice_list_view_out_of_order() -> Result<()> {
1327 let values: ArrayRef = Arc::new(Int32Array::from(vec![1, 2, 3, 4, 5]));
1328 let offsets = ScalarBuffer::from(vec![3, 1, 0]);
1329 let sizes = ScalarBuffer::from(vec![2, 2, 1]);
1330 let field = Arc::new(Field::new("item", DataType::Int32, true));
1331 let array = ListViewArray::new(field, offsets, sizes, values, None);
1332 assert_eq!(
1333 list_view_values(&array),
1334 vec![vec![4, 5], vec![2, 3], vec![1]]
1335 );
1336
1337 let from = Int64Array::from(vec![2, 2, 2]);
1338 let to = Int64Array::from(vec![1, 1, 1]);
1339 let stride = Int64Array::from(vec![-1, -1, -1]);
1340
1341 let result =
1342 general_list_view_array_slice::<i32>(&array, &from, &to, Some(&stride))?;
1343 let result = result.as_ref().as_list_view::<i32>();
1344
1345 assert_eq!(
1346 list_view_values(result),
1347 vec![vec![5, 4], vec![3, 2], vec![]]
1348 );
1349 Ok(())
1350 }
1351
1352 #[test]
1353 fn test_array_slice_list_view_with_nulls() -> Result<()> {
1354 let values: ArrayRef = Arc::new(Int32Array::from(vec![
1355 Some(1),
1356 None,
1357 Some(3),
1358 Some(4),
1359 Some(5),
1360 ]));
1361 let offsets = ScalarBuffer::from(vec![0, 2, 5]);
1362 let sizes = ScalarBuffer::from(vec![2, 3, 0]);
1363 let field = Arc::new(Field::new("item", DataType::Int32, true));
1364 let array = ListViewArray::new(field, offsets, sizes, values, None);
1365
1366 let from = Int64Array::from(vec![1, 1, 1]);
1367 let to = Int64Array::from(vec![2, 2, 1]);
1368
1369 let result = general_list_view_array_slice::<i32>(&array, &from, &to, None)?;
1370 let result = result.as_ref().as_list_view::<i32>();
1371
1372 let actual: Vec<Vec<Option<i32>>> = (0..result.len())
1373 .map(|i| {
1374 result
1375 .value(i)
1376 .as_any()
1377 .downcast_ref::<Int32Array>()
1378 .unwrap()
1379 .iter()
1380 .collect()
1381 })
1382 .collect();
1383
1384 assert_eq!(
1385 actual,
1386 vec![vec![Some(1), None], vec![Some(3), Some(4)], Vec::new(),]
1387 );
1388
1389 let stride_with_null = Int64Array::from(vec![Some(1), None, Some(1)]);
1391 let result = general_list_view_array_slice::<i32>(
1392 &array,
1393 &from,
1394 &to,
1395 Some(&stride_with_null),
1396 )?;
1397 let result = result.as_ref().as_list_view::<i32>();
1398
1399 assert!(!result.is_null(0)); assert!(result.is_null(1)); assert!(!result.is_null(2)); let first_row: Vec<Option<i32>> = result
1407 .value(0)
1408 .as_any()
1409 .downcast_ref::<Int32Array>()
1410 .unwrap()
1411 .iter()
1412 .collect();
1413 assert_eq!(first_row, vec![Some(1), None]);
1414
1415 Ok(())
1416 }
1417}