datafusion_functions/core/
struct.rsuse arrow::array::{ArrayRef, StructArray};
use arrow::datatypes::{DataType, Field, Fields};
use datafusion_common::{exec_err, Result};
use datafusion_expr::scalar_doc_sections::DOC_SECTION_STRUCT;
use datafusion_expr::{ColumnarValue, Documentation};
use datafusion_expr::{ScalarUDFImpl, Signature, Volatility};
use std::any::Any;
use std::sync::{Arc, OnceLock};
fn array_struct(args: &[ArrayRef]) -> Result<ArrayRef> {
if args.is_empty() {
return exec_err!("struct requires at least one argument");
}
let fields = args
.iter()
.enumerate()
.map(|(i, arg)| {
let field_name = format!("c{i}");
Ok(Arc::new(Field::new(
field_name.as_str(),
arg.data_type().clone(),
true,
)))
})
.collect::<Result<Vec<_>>>()?
.into();
let arrays = args.to_vec();
Ok(Arc::new(StructArray::new(fields, arrays, None)))
}
fn struct_expr(args: &[ColumnarValue]) -> Result<ColumnarValue> {
let arrays = ColumnarValue::values_to_arrays(args)?;
Ok(ColumnarValue::Array(array_struct(arrays.as_slice())?))
}
#[derive(Debug)]
pub struct StructFunc {
signature: Signature,
aliases: Vec<String>,
}
impl Default for StructFunc {
fn default() -> Self {
Self::new()
}
}
impl StructFunc {
pub fn new() -> Self {
Self {
signature: Signature::variadic_any(Volatility::Immutable),
aliases: vec![String::from("row")],
}
}
}
impl ScalarUDFImpl for StructFunc {
fn as_any(&self) -> &dyn Any {
self
}
fn name(&self) -> &str {
"struct"
}
fn aliases(&self) -> &[String] {
&self.aliases
}
fn signature(&self) -> &Signature {
&self.signature
}
fn return_type(&self, arg_types: &[DataType]) -> Result<DataType> {
let return_fields = arg_types
.iter()
.enumerate()
.map(|(pos, dt)| Field::new(format!("c{pos}"), dt.clone(), true))
.collect::<Vec<Field>>();
Ok(DataType::Struct(Fields::from(return_fields)))
}
fn invoke(&self, args: &[ColumnarValue]) -> Result<ColumnarValue> {
struct_expr(args)
}
fn documentation(&self) -> Option<&Documentation> {
Some(get_struct_doc())
}
}
static DOCUMENTATION: OnceLock<Documentation> = OnceLock::new();
fn get_struct_doc() -> &'static Documentation {
DOCUMENTATION.get_or_init(|| {
Documentation::builder()
.with_doc_section(DOC_SECTION_STRUCT)
.with_description("Returns an Arrow struct using the specified input expressions optionally named.
Fields in the returned struct use the optional name or the `cN` naming convention.
For example: `c0`, `c1`, `c2`, etc.")
.with_syntax_example("struct(expression1[, ..., expression_n])")
.with_sql_example(r#"For example, this query converts two columns `a` and `b` to a single column with
a struct type of fields `field_a` and `c1`:
```sql
> select * from t;
+---+---+
| a | b |
+---+---+
| 1 | 2 |
| 3 | 4 |
+---+---+
-- use default names `c0`, `c1`
> select struct(a, b) from t;
+-----------------+
| struct(t.a,t.b) |
+-----------------+
| {c0: 1, c1: 2} |
| {c0: 3, c1: 4} |
+-----------------+
-- name the first field `field_a`
select struct(a as field_a, b) from t;
+--------------------------------------------------+
| named_struct(Utf8("field_a"),t.a,Utf8("c1"),t.b) |
+--------------------------------------------------+
| {field_a: 1, c1: 2} |
| {field_a: 3, c1: 4} |
+--------------------------------------------------+
```
"#)
.with_argument(
"expression1, expression_n",
"Expression to include in the output struct. Can be a constant, column, or function, any combination of arithmetic or string operators.")
.build()
.unwrap()
})
}