1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
use arrow::array::{new_null_array, Array, BooleanArray};
use arrow::compute::kernels::zip::zip;
use arrow::compute::{and, is_not_null, is_null};
use datafusion_common::{DataFusionError, Result};
use datafusion_expr::ColumnarValue;
pub fn coalesce(args: &[ColumnarValue]) -> Result<ColumnarValue> {
if args.is_empty() {
return Err(DataFusionError::Internal(format!(
"coalesce was called with {} arguments. It requires at least 1.",
args.len()
)));
}
let return_type = args[0].data_type();
let mut return_array = args.iter().filter_map(|x| match x {
ColumnarValue::Array(array) => Some(array.len()),
_ => None,
});
if let Some(size) = return_array.next() {
let mut current_value = new_null_array(&return_type, size);
let mut remainder = BooleanArray::from(vec![true; size]);
for arg in args {
match arg {
ColumnarValue::Array(ref array) => {
let to_apply = and(&remainder, &is_not_null(array.as_ref())?)?;
current_value = zip(&to_apply, array, current_value.as_ref())?;
remainder = and(&remainder, &is_null(array)?)?;
}
ColumnarValue::Scalar(value) => {
if value.is_null() {
continue;
} else {
let last_value = value.to_array_of_size(size);
current_value =
zip(&remainder, &last_value, current_value.as_ref())?;
break;
}
}
}
if remainder.iter().all(|x| x == Some(false)) {
break;
}
}
Ok(ColumnarValue::Array(current_value))
} else {
let result = args
.iter()
.filter_map(|x| match x {
ColumnarValue::Scalar(s) if !s.is_null() => Some(x.clone()),
_ => None,
})
.next()
.unwrap_or_else(|| args[0].clone());
Ok(result)
}
}