use std::any::Any;
use arrow::compute::kernels::length::length;
use arrow::datatypes::DataType;
use datafusion_common::{exec_err, Result, ScalarValue};
use datafusion_expr::{ColumnarValue, Volatility};
use datafusion_expr::{ScalarUDFImpl, Signature};
use crate::utils::utf8_to_int_type;
#[derive(Debug)]
pub struct OctetLengthFunc {
signature: Signature,
}
impl Default for OctetLengthFunc {
fn default() -> Self {
Self::new()
}
}
impl OctetLengthFunc {
pub fn new() -> Self {
use DataType::*;
Self {
signature: Signature::uniform(
1,
vec![Utf8, LargeUtf8],
Volatility::Immutable,
),
}
}
}
impl ScalarUDFImpl for OctetLengthFunc {
fn as_any(&self) -> &dyn Any {
self
}
fn name(&self) -> &str {
"octet_length"
}
fn signature(&self) -> &Signature {
&self.signature
}
fn return_type(&self, arg_types: &[DataType]) -> Result<DataType> {
utf8_to_int_type(&arg_types[0], "octet_length")
}
fn invoke(&self, args: &[ColumnarValue]) -> Result<ColumnarValue> {
if args.len() != 1 {
return exec_err!(
"octet_length function requires 1 argument, got {}",
args.len()
);
}
match &args[0] {
ColumnarValue::Array(v) => Ok(ColumnarValue::Array(length(v.as_ref())?)),
ColumnarValue::Scalar(v) => match v {
ScalarValue::Utf8(v) => Ok(ColumnarValue::Scalar(ScalarValue::Int32(
v.as_ref().map(|x| x.len() as i32),
))),
ScalarValue::LargeUtf8(v) => Ok(ColumnarValue::Scalar(
ScalarValue::Int64(v.as_ref().map(|x| x.len() as i64)),
)),
_ => unreachable!(),
},
}
}
}
#[cfg(test)]
mod tests {
use std::sync::Arc;
use arrow::array::{Array, Int32Array, StringArray};
use arrow::datatypes::DataType::Int32;
use datafusion_common::ScalarValue;
use datafusion_common::{exec_err, Result};
use datafusion_expr::{ColumnarValue, ScalarUDFImpl};
use crate::string::octet_length::OctetLengthFunc;
use crate::utils::test::test_function;
#[test]
fn test_functions() -> Result<()> {
test_function!(
OctetLengthFunc::new(),
&[ColumnarValue::Scalar(ScalarValue::Int32(Some(12)))],
exec_err!(
"The OCTET_LENGTH function can only accept strings, but got Int32."
),
i32,
Int32,
Int32Array
);
test_function!(
OctetLengthFunc::new(),
&[ColumnarValue::Array(Arc::new(StringArray::from(vec![
String::from("chars"),
String::from("chars2"),
])))],
Ok(Some(5)),
i32,
Int32,
Int32Array
);
test_function!(
OctetLengthFunc::new(),
&[
ColumnarValue::Scalar(ScalarValue::Utf8(Some(String::from("chars")))),
ColumnarValue::Scalar(ScalarValue::Utf8(Some(String::from("chars"))))
],
exec_err!("octet_length function requires 1 argument, got 2"),
i32,
Int32,
Int32Array
);
test_function!(
OctetLengthFunc::new(),
&[ColumnarValue::Scalar(ScalarValue::Utf8(Some(
String::from("chars")
)))],
Ok(Some(5)),
i32,
Int32,
Int32Array
);
test_function!(
OctetLengthFunc::new(),
&[ColumnarValue::Scalar(ScalarValue::Utf8(Some(
String::from("josé")
)))],
Ok(Some(5)),
i32,
Int32,
Int32Array
);
test_function!(
OctetLengthFunc::new(),
&[ColumnarValue::Scalar(ScalarValue::Utf8(Some(
String::from("")
)))],
Ok(Some(0)),
i32,
Int32,
Int32Array
);
test_function!(
OctetLengthFunc::new(),
&[ColumnarValue::Scalar(ScalarValue::Utf8(None))],
Ok(None),
i32,
Int32,
Int32Array
);
Ok(())
}
}