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
use arrow::array::*;
use arrow::datatypes::{DataType, Field};
use datafusion_common::{DataFusionError, Result};
use datafusion_expr::ColumnarValue;
use std::sync::Arc;
fn array_struct(args: &[ArrayRef]) -> Result<ArrayRef> {
if args.is_empty() {
return Err(DataFusionError::Internal(
"struct requires at least one argument".to_string(),
));
}
let vec: Vec<_> = args
.iter()
.enumerate()
.map(|(i, arg)| -> Result<(Field, ArrayRef)> {
let field_name = format!("c{i}");
match arg.data_type() {
DataType::Utf8
| DataType::LargeUtf8
| DataType::Boolean
| DataType::Float32
| DataType::Float64
| DataType::Int8
| DataType::Int16
| DataType::Int32
| DataType::Int64
| DataType::UInt8
| DataType::UInt16
| DataType::UInt32
| DataType::UInt64 => Ok((
Field::new(field_name.as_str(), arg.data_type().clone(), true),
arg.clone(),
)),
data_type => Err(DataFusionError::NotImplemented(format!(
"Struct is not implemented for type '{data_type:?}'."
))),
}
})
.collect::<Result<Vec<_>>>()?;
Ok(Arc::new(StructArray::from(vec)))
}
pub fn struct_expr(values: &[ColumnarValue]) -> Result<ColumnarValue> {
let arrays: Vec<ArrayRef> = values
.iter()
.map(|x| match x {
ColumnarValue::Array(array) => array.clone(),
ColumnarValue::Scalar(scalar) => scalar.to_array().clone(),
})
.collect();
Ok(ColumnarValue::Array(array_struct(arrays.as_slice())?))
}