Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fix coalesce, struct and named_strct expr_fn function to take multiple arguments #10321

Merged
merged 3 commits into from
May 7, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
177 changes: 176 additions & 1 deletion datafusion/core/tests/dataframe/dataframe_functions.rs
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@ use datafusion::error::Result;
use datafusion::prelude::*;

use datafusion::assert_batches_eq;
use datafusion_common::DFSchema;
use datafusion_common::{DFSchema, ScalarValue};
use datafusion_expr::expr::Alias;
use datafusion_expr::ExprSchemable;

Expand Down Expand Up @@ -161,6 +161,181 @@ async fn test_fn_btrim_with_chars() -> Result<()> {
Ok(())
}

#[tokio::test]
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

While adding a test for coalesce I realized the other functions in core weren't covered so I added new tests for them too (except for get_field which needs a map / struct and I was too lazy to add one to the test fixture

async fn test_fn_nullif() -> Result<()> {
let expr = nullif(col("a"), lit("abcDEF"));

let expected = [
"+-------------------------------+",
"| nullif(test.a,Utf8(\"abcDEF\")) |",
"+-------------------------------+",
"| |",
"| abc123 |",
"| CBAdef |",
"| 123AbcDef |",
"+-------------------------------+",
];

assert_fn_batches!(expr, expected);

Ok(())
}

#[tokio::test]
async fn test_fn_arrow_cast() -> Result<()> {
let expr = arrow_typeof(arrow_cast(col("b"), lit("Float64")));

let expected = [
"+--------------------------------------------------+",
"| arrow_typeof(arrow_cast(test.b,Utf8(\"Float64\"))) |",
"+--------------------------------------------------+",
"| Float64 |",
"| Float64 |",
"| Float64 |",
"| Float64 |",
"+--------------------------------------------------+",
];

assert_fn_batches!(expr, expected);

Ok(())
}

#[tokio::test]
async fn test_nvl() -> Result<()> {
let lit_null = lit(ScalarValue::Utf8(None));
// nvl(CASE WHEN a = 'abcDEF' THEN NULL ELSE a END, 'TURNED_NULL')
let expr = nvl(
when(col("a").eq(lit("abcDEF")), lit_null)
.otherwise(col("a"))
.unwrap(),
lit("TURNED_NULL"),
)
.alias("nvl_expr");

let expected = [
"+-------------+",
"| nvl_expr |",
"+-------------+",
"| TURNED_NULL |",
"| abc123 |",
"| CBAdef |",
"| 123AbcDef |",
"+-------------+",
];

assert_fn_batches!(expr, expected);

Ok(())
}
#[tokio::test]
async fn test_nvl2() -> Result<()> {
let lit_null = lit(ScalarValue::Utf8(None));
// nvl2(CASE WHEN a = 'abcDEF' THEN NULL ELSE a END, 'NON_NUll', 'TURNED_NULL')
let expr = nvl2(
when(col("a").eq(lit("abcDEF")), lit_null)
.otherwise(col("a"))
.unwrap(),
lit("NON_NULL"),
lit("TURNED_NULL"),
)
.alias("nvl2_expr");

let expected = [
"+-------------+",
"| nvl2_expr |",
"+-------------+",
"| TURNED_NULL |",
"| NON_NULL |",
"| NON_NULL |",
"| NON_NULL |",
"+-------------+",
];

assert_fn_batches!(expr, expected);

Ok(())
}
#[tokio::test]
async fn test_fn_arrow_typeof() -> Result<()> {
let expr = arrow_typeof(col("l"));

let expected = [
"+------------------------------------------------------------------------------------------------------------------+",
"| arrow_typeof(test.l) |",
"+------------------------------------------------------------------------------------------------------------------+",
"| List(Field { name: \"item\", data_type: Int32, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }) |",
"| List(Field { name: \"item\", data_type: Int32, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }) |",
"| List(Field { name: \"item\", data_type: Int32, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }) |",
"| List(Field { name: \"item\", data_type: Int32, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }) |",
"+------------------------------------------------------------------------------------------------------------------+",
];

assert_fn_batches!(expr, expected);

Ok(())
}

#[tokio::test]
async fn test_fn_struct() -> Result<()> {
let expr = r#struct(vec![col("a"), col("b")]);

let expected = [
"+--------------------------+",
"| struct(test.a,test.b) |",
"+--------------------------+",
"| {c0: abcDEF, c1: 1} |",
"| {c0: abc123, c1: 10} |",
"| {c0: CBAdef, c1: 10} |",
"| {c0: 123AbcDef, c1: 100} |",
"+--------------------------+",
];

assert_fn_batches!(expr, expected);

Ok(())
}

#[tokio::test]
async fn test_fn_named_struct() -> Result<()> {
let expr = named_struct(vec![lit("column_a"), col("a"), lit("column_b"), col("b")]);

let expected = [
"+---------------------------------------------------------------+",
"| named_struct(Utf8(\"column_a\"),test.a,Utf8(\"column_b\"),test.b) |",
"+---------------------------------------------------------------+",
"| {column_a: abcDEF, column_b: 1} |",
"| {column_a: abc123, column_b: 10} |",
"| {column_a: CBAdef, column_b: 10} |",
"| {column_a: 123AbcDef, column_b: 100} |",
"+---------------------------------------------------------------+",
];

assert_fn_batches!(expr, expected);

Ok(())
}

#[tokio::test]
async fn test_fn_coalesce() -> Result<()> {
let expr = coalesce(vec![lit(ScalarValue::Utf8(None)), lit("ab")]);

let expected = [
"+---------------------------------+",
"| coalesce(Utf8(NULL),Utf8(\"ab\")) |",
"+---------------------------------+",
"| ab |",
"| ab |",
"| ab |",
"| ab |",
"+---------------------------------+",
];

assert_fn_batches!(expr, expected);

Ok(())
}

#[tokio::test]
async fn test_fn_approx_median() -> Result<()> {
let expr = approx_median(col("b"));
Expand Down
79 changes: 68 additions & 11 deletions datafusion/functions/src/core/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,9 @@

//! "core" DataFusion functions

use datafusion_expr::ScalarUDF;
use std::sync::Arc;

pub mod arrow_cast;
pub mod arrowtypeof;
pub mod coalesce;
Expand All @@ -39,14 +42,68 @@ make_udf_function!(getfield::GetFieldFunc, GET_FIELD, get_field);
make_udf_function!(coalesce::CoalesceFunc, COALESCE, coalesce);

// Export the functions out of this package, both as expr_fn as well as a list of functions
export_functions!(
(nullif, arg_1 arg_2, "returns NULL if value1 equals value2; otherwise it returns value1. This can be used to perform the inverse operation of the COALESCE expression."),
(arrow_cast, arg_1 arg_2, "returns arg_1 cast to the `arrow_type` given the second argument. This can be used to cast to a specific `arrow_type`."),
(nvl, arg_1 arg_2, "returns value2 if value1 is NULL; otherwise it returns value1"),
(nvl2, arg_1 arg_2 arg_3, "Returns value2 if value1 is not NULL; otherwise, it returns value3."),
(arrow_typeof, arg_1, "Returns the Arrow type of the input expression."),
(r#struct, args, "Returns a struct with the given arguments"),
(named_struct, args, "Returns a struct with the given names and arguments pairs"),
(get_field, arg_1 arg_2, "Returns the value of the field with the given name from the struct"),
(coalesce, args, "Returns `coalesce(args...)`, which evaluates to the value of the first expr which is not NULL")
);
pub mod expr_fn {
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It is not possible to create an expr_fn that takes a Vec<Expr> using the export_functions macro, so follow the model @Omega359 used in the math module

pub mod expr_fn {
use datafusion_expr::Expr;
#[doc = "returns the absolute value of a given number"]
pub fn abs(num: Expr) -> Expr {
super::abs().call(vec![num])
}
#[doc = "returns the arc cosine or inverse cosine of a number"]
pub fn acos(num: Expr) -> Expr {
super::acos().call(vec![num])
}
#[doc = "returns inverse hyperbolic cosine"]
pub fn acosh(num: Expr) -> Expr {
super::acosh().call(vec![num])
}
#[doc = "returns the arc sine or inverse sine of a number"]
pub fn asin(num: Expr) -> Expr {
super::asin().call(vec![num])
}
#[doc = "returns inverse hyperbolic sine"]
pub fn asinh(num: Expr) -> Expr {
super::asinh().call(vec![num])
}
#[doc = "returns inverse tangent"]
pub fn atan(num: Expr) -> Expr {
super::atan().call(vec![num])
}
#[doc = "returns inverse tangent of a division given in the argument"]
pub fn atan2(y: Expr, x: Expr) -> Expr {
super::atan2().call(vec![y, x])
}
#[doc = "returns inverse hyperbolic tangent"]
pub fn atanh(num: Expr) -> Expr {
super::atanh().call(vec![num])
}
#[doc = "cube root of a number"]
pub fn cbrt(num: Expr) -> Expr {
super::cbrt().call(vec![num])
}
#[doc = "nearest integer greater than or equal to argument"]
pub fn ceil(num: Expr) -> Expr {
super::ceil().call(vec![num])
}
#[doc = "cosine"]
pub fn cos(num: Expr) -> Expr {
super::cos().call(vec![num])
}
#[doc = "hyperbolic cosine"]
pub fn cosh(num: Expr) -> Expr {
super::cosh().call(vec![num])
}
#[doc = "cotangent of a number"]
pub fn cot(num: Expr) -> Expr {
super::cot().call(vec![num])
}
#[doc = "converts radians to degrees"]
pub fn degrees(num: Expr) -> Expr {
super::degrees().call(vec![num])
}
#[doc = "exponential"]
pub fn exp(num: Expr) -> Expr {
super::exp().call(vec![num])
}
#[doc = "factorial"]
pub fn factorial(num: Expr) -> Expr {
super::factorial().call(vec![num])
}
#[doc = "nearest integer less than or equal to argument"]
pub fn floor(num: Expr) -> Expr {
super::floor().call(vec![num])
}
#[doc = "greatest common divisor"]
pub fn gcd(x: Expr, y: Expr) -> Expr {
super::gcd().call(vec![x, y])
}
#[doc = "returns true if a given number is +NaN or -NaN otherwise returns false"]
pub fn isnan(num: Expr) -> Expr {
super::isnan().call(vec![num])
}
#[doc = "returns true if a given number is +0.0 or -0.0 otherwise returns false"]
pub fn iszero(num: Expr) -> Expr {
super::iszero().call(vec![num])
}
#[doc = "least common multiple"]
pub fn lcm(x: Expr, y: Expr) -> Expr {
super::lcm().call(vec![x, y])
}
#[doc = "natural logarithm (base e) of a number"]
pub fn ln(num: Expr) -> Expr {
super::ln().call(vec![num])
}
#[doc = "logarithm of a number for a particular `base`"]
pub fn log(base: Expr, num: Expr) -> Expr {
super::log().call(vec![base, num])
}
#[doc = "base 2 logarithm of a number"]
pub fn log2(num: Expr) -> Expr {
super::log2().call(vec![num])
}
#[doc = "base 10 logarithm of a number"]
pub fn log10(num: Expr) -> Expr {
super::log10().call(vec![num])
}
#[doc = "returns x if x is not NaN otherwise returns y"]
pub fn nanvl(x: Expr, y: Expr) -> Expr {
super::nanvl().call(vec![x, y])
}
#[doc = "Returns an approximate value of π"]
pub fn pi() -> Expr {
super::pi().call(vec![])
}
#[doc = "`base` raised to the power of `exponent`"]
pub fn power(base: Expr, exponent: Expr) -> Expr {
super::power().call(vec![base, exponent])
}
#[doc = "converts degrees to radians"]
pub fn radians(num: Expr) -> Expr {
super::radians().call(vec![num])
}
#[doc = "Returns a random value in the range 0.0 <= x < 1.0"]
pub fn random() -> Expr {
super::random().call(vec![])
}
#[doc = "round to nearest integer"]
pub fn round(args: Vec<Expr>) -> Expr {
super::round().call(args)
}
#[doc = "sign of the argument (-1, 0, +1)"]
pub fn signum(num: Expr) -> Expr {
super::signum().call(vec![num])
}
#[doc = "sine"]
pub fn sin(num: Expr) -> Expr {
super::sin().call(vec![num])
}
#[doc = "hyperbolic sine"]
pub fn sinh(num: Expr) -> Expr {
super::sinh().call(vec![num])
}
#[doc = "square root of a number"]
pub fn sqrt(num: Expr) -> Expr {
super::sqrt().call(vec![num])
}
#[doc = "returns the tangent of a number"]
pub fn tan(num: Expr) -> Expr {
super::tan().call(vec![num])
}
#[doc = "returns the hyperbolic tangent of a number"]
pub fn tanh(num: Expr) -> Expr {
super::tanh().call(vec![num])
}
#[doc = "truncate toward zero, with optional precision"]
pub fn trunc(args: Vec<Expr>) -> Expr {
super::trunc().call(args)
}
}

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think it is possible to take Vec like what functions-array macro does

($UDF:ty, $EXPR_FN:ident, $DOC:expr , $SCALAR_UDF_FN:ident) => {
        paste::paste! {
            // "fluent expr_fn" style function
            #[doc = $DOC]
            pub fn $EXPR_FN(arg: Vec<Expr>) -> Expr {
                Expr::ScalarFunction(ScalarFunction::new_udf(
                    $SCALAR_UDF_FN(),
                    arg,
                ))
            }

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What I could't figure out how to do was make the macro take both syntaxes

It would have to look something like

export_functions!(
    // create a function with arg_1 and arg_2 Expr arguments
    (arrow_cast, arg_1 arg_2, "returns arg_1 cast to the `arrow_type` given the second argument. This can be used to cast to a specific `arrow_type`."),
   /// create a function with a single  Vec<Expr> arg
    (coalesce, args, "Returns `coalesce(args...)`, which evaluates to the value of the first expr which is not NULL")
...
}

the macro definition is

macro_rules! export_functions {
($(($FUNC:ident, $($arg:ident)*, $DOC:expr)),*) => {
pub mod expr_fn {
$(
#[doc = $DOC]
/// Return $name(arg)
pub fn $FUNC($($arg: datafusion_expr::Expr),*) -> datafusion_expr::Expr {
super::$FUNC().call(vec![$($arg),*],)
}
)*
}
/// Return a list of all functions in this package
pub fn functions() -> Vec<std::sync::Arc<datafusion_expr::ScalarUDF>> {
vec![
$(
$FUNC(),
)*
]
}
};
}

there may be some way to do this in rust, but I could't figure it out

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I spent a whole day trying to make it work as well and got lost in the depths of macro programming. I'm not sure it's worth the effort of trying to get it to work tbh - the alternative is not exactly hard work.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

macro_rules! make_udf_function {
($UDF:ty, $EXPR_FN:ident, $($arg:ident)*, $DOC:expr , $SCALAR_UDF_FN:ident) => {
paste::paste! {
// "fluent expr_fn" style function
#[doc = $DOC]
pub fn $EXPR_FN($($arg: Expr),*) -> Expr {
Expr::ScalarFunction(ScalarFunction::new_udf(
$SCALAR_UDF_FN(),
vec![$($arg),*],
))
}
/// Singleton instance of [`$UDF`], ensures the UDF is only created once
/// named STATIC_$(UDF). For example `STATIC_ArrayToString`
#[allow(non_upper_case_globals)]
static [< STATIC_ $UDF >]: std::sync::OnceLock<std::sync::Arc<datafusion_expr::ScalarUDF>> =
std::sync::OnceLock::new();
/// ScalarFunction that returns a [`ScalarUDF`] for [`$UDF`]
///
/// [`ScalarUDF`]: datafusion_expr::ScalarUDF
pub fn $SCALAR_UDF_FN() -> std::sync::Arc<datafusion_expr::ScalarUDF> {
[< STATIC_ $UDF >]
.get_or_init(|| {
std::sync::Arc::new(datafusion_expr::ScalarUDF::new_from_impl(
<$UDF>::new(),
))
})
.clone()
}
}
};
($UDF:ty, $EXPR_FN:ident, $DOC:expr , $SCALAR_UDF_FN:ident) => {
paste::paste! {
// "fluent expr_fn" style function
#[doc = $DOC]
pub fn $EXPR_FN(arg: Vec<Expr>) -> Expr {
Expr::ScalarFunction(ScalarFunction::new_udf(
$SCALAR_UDF_FN(),
arg,
))
}
/// Singleton instance of [`$UDF`], ensures the UDF is only created once
/// named STATIC_$(UDF). For example `STATIC_ArrayToString`
#[allow(non_upper_case_globals)]
static [< STATIC_ $UDF >]: std::sync::OnceLock<std::sync::Arc<datafusion_expr::ScalarUDF>> =
std::sync::OnceLock::new();
/// ScalarFunction that returns a [`ScalarUDF`] for [`$UDF`]
///
/// [`ScalarUDF`]: datafusion_expr::ScalarUDF
pub fn $SCALAR_UDF_FN() -> std::sync::Arc<datafusion_expr::ScalarUDF> {
[< STATIC_ $UDF >]
.get_or_init(|| {
std::sync::Arc::new(datafusion_expr::ScalarUDF::new_from_impl(
<$UDF>::new(),
))
})
.clone()
}
}
};
}

In array macro, we support two syntaxes, which is surprisingly easy

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I was thinking of alamb#19.

However, it does not reuse the macro export_functions that accepts multiple functions with one macro call.

export_functions_single accepts a single function with each macro call.

Since this requires introducing another macro, it is also not an ideal solution.

I'm fine to move on with the current PR.

Copy link
Contributor

@jayzhan211 jayzhan211 May 3, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

https://users.rust-lang.org/t/macro-repetition-with-multiple-rules/110816/2?u=jayzhan

I got a probably better solution (I had not tried it) from rust forum. To anyone that is interested in

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thank you @jayzhan211 -- I will give it a try

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I couldn't make this work (though I didn't try very hard) so I filed #10397 as a follow on task for this

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we can merge this first and release 38.

use datafusion_expr::Expr;

/// returns NULL if value1 equals value2; otherwise it returns value1. This
/// can be used to perform the inverse operation of the COALESCE expression
pub fn nullif(arg1: Expr, arg2: Expr) -> Expr {
super::nullif().call(vec![arg1, arg2])
}

/// returns value1 cast to the `arrow_type` given the second argument. This
/// can be used to cast to a specific `arrow_type`.
pub fn arrow_cast(arg1: Expr, arg2: Expr) -> Expr {
super::arrow_cast().call(vec![arg1, arg2])
}

/// Returns value2 if value1 is NULL; otherwise it returns value1
pub fn nvl(arg1: Expr, arg2: Expr) -> Expr {
super::nvl().call(vec![arg1, arg2])
}

/// Returns value2 if value1 is not NULL; otherwise, it returns value3.
pub fn nvl2(arg1: Expr, arg2: Expr, arg3: Expr) -> Expr {
super::nvl2().call(vec![arg1, arg2, arg3])
}

/// Returns the Arrow type of the input expression.
pub fn arrow_typeof(arg1: Expr) -> Expr {
super::arrow_typeof().call(vec![arg1])
}

/// Returns a struct with the given arguments
pub fn r#struct(args: Vec<Expr>) -> Expr {
super::r#struct().call(args)
}

/// Returns a struct with the given names and arguments pairs
pub fn named_struct(args: Vec<Expr>) -> Expr {
super::named_struct().call(args)
}

/// Returns the value of the field with the given name from the struct
pub fn get_field(arg1: Expr, arg2: Expr) -> Expr {
super::get_field().call(vec![arg1, arg2])
}

/// Returns `coalesce(args...)`, which evaluates to the value of the first expr which is not NULL
pub fn coalesce(args: Vec<Expr>) -> Expr {
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is the actual fix, though I suspect that struct and named_struct are also affected

super::coalesce().call(args)
}
}

/// Return a list of all functions in this package
pub fn functions() -> Vec<Arc<ScalarUDF>> {
vec![
nullif(),
arrow_cast(),
nvl(),
nvl2(),
arrow_typeof(),
r#struct(),
named_struct(),
get_field(),
coalesce(),
]
}