Fix functions with Volatility::Volatile and parameters #13001

agscpp · 2024-10-18T12:56:20Z

Which issue does this PR close?

Rationale for this change

These changes will allow functions to be implemented that produce a unique result for each call given the same input.

alamb

Thank you @agscpp -- this makes a lot of sense and I think is quite close. I left some API comments for your consideration. Let me know!

alamb · 2024-10-18T19:21:07Z

datafusion/expr/src/udf.rs

+        not_impl_err!(
+            "Function {} does not implement invoke_batch but called",
+            self.name()
+        )


🤔 It seems like the ideal outcome would be for all ScalarUDFs to implement this method (as it covers invoke, invoke_no_args as well).

Would you be open to changing this so it uses a default implementation like this?

Suggested change

not_impl_err!(

"Function {} does not implement invoke_batch but called",

self.name()

)

if _args.empty() {

self.invoke_no_args(number_rows)

} else {

self.invoke(args)

}

Then the function implementation could decide what to do with that information

alamb · 2024-10-18T19:23:19Z

datafusion/core/tests/user_defined/user_defined_scalar_functions.rs

+            "+-----------+", //
+            "| str       |", //
+            "+-----------+", //
+            "| 1. test_1 |", //


what is the meaning of the trailing // ?

Also, it seems like the indexes repeat (multiple with 1) imply invoke is run multiple times - perhaps we could set target_partitions to 1 on the SessionContext so the data wasn't repartitioned?

I myself did not understand what the // are for, but I left it because this style was used above.

the invoke_batch method is called once. In the logic of the function itself, I wrote the indexing of the module by the contents of the lines.

1. test_1 2. test_1 3. test_1 4. test_1

1. test_2 2. test_2 3. test_2

datafusion/core/tests/user_defined/user_defined_scalar_functions.rs

alamb · 2024-10-18T19:25:48Z

datafusion/physical-expr/src/scalar_function.rs

@@ -143,7 +143,10 @@ impl PhysicalExpr for ScalarFunctionExpr {
        // evaluate the function
        let output = match self.args.is_empty() {
            true => self.fun.invoke_no_args(batch.num_rows()),
-            false => self.fun.invoke(&inputs),
+            false => match self.fun.signature().volatility {


If you modified invoke_batch as above, we could change this code to simply call self.fun.invoke_batch() always

alamb

Thank you @agscpp -- this looks great to me

alamb · 2024-10-21T16:27:47Z

datafusion/expr/src/udf.rs

@@ -467,7 +478,28 @@ pub trait ScalarUDFImpl: Debug + Send + Sync {
    /// to arrays, which will likely be simpler code, but be slower.
    ///
    /// [invoke_no_args]: ScalarUDFImpl::invoke_no_args
-    fn invoke(&self, _args: &[ColumnarValue]) -> Result<ColumnarValue>;
+    fn invoke(&self, _args: &[ColumnarValue]) -> Result<ColumnarValue> {


Nice! -- as a follow on PR I think we should deprecate the other two functions (invoke_no_args and invoke) telling people to use invoke instead

Is this ok with you @jayzhan211 ?

as a follow on PR I think we should deprecate the other two functions (invoke_no_args and invoke) telling people to use invoke instead

did you mean invoke_batch?

yes, it would be great to have only one invoke entry-point

alamb · 2024-10-21T16:28:47Z

datafusion/core/tests/user_defined/user_defined_scalar_functions.rs

+        assert_batches_eq!(expected, &result);
+
+        let result =
+            plan_and_collect(&ctx, "select add_index_to_string('test') AS str from t") // with fixed function parameters


findepi · 2024-10-22T10:42:28Z

datafusion/expr/src/udf.rs

@@ -467,7 +478,28 @@ pub trait ScalarUDFImpl: Debug + Send + Sync {
    /// to arrays, which will likely be simpler code, but be slower.
    ///
    /// [invoke_no_args]: ScalarUDFImpl::invoke_no_args
-    fn invoke(&self, _args: &[ColumnarValue]) -> Result<ColumnarValue>;
+    fn invoke(&self, _args: &[ColumnarValue]) -> Result<ColumnarValue> {


as a follow on PR I think we should deprecate the other two functions (invoke_no_args and invoke) telling people to use invoke instead

did you mean invoke_batch?

yes, it would be great to have only one invoke entry-point

findepi · 2024-10-22T10:43:12Z

datafusion/expr/src/udf.rs

+        _args: &[ColumnarValue],
+        _number_rows: usize,


These are not unused, let's remove leading _ from arg names

findepi · 2024-10-22T10:44:02Z

datafusion/expr/src/udf.rs

+    /// The function should be used for signatures with [`datafusion_expr_common::signature::Volatility::Volatile`]
+    /// and with arguments.


Only for these?

if yes => the function should be named appropriate invoke_volatile)
if no => remove the comment

I removed this comment. This function is always called in the current implementation.

findepi · 2024-10-22T10:45:59Z

datafusion/core/tests/user_defined/user_defined_scalar_functions.rs

+    }
+    {


Nit: consider separating test cases into separate test functions, this would given them descriptive names

These tests are very similar, so they don't break down into two parts well.

findepi · 2024-10-22T10:47:18Z

datafusion/core/tests/user_defined/user_defined_scalar_functions.rs

+/// Volatile UDF that should be append a different value to each row
+struct AddIndexToStringScalarUDF {


The volatility is important, let's reflect it in the function name -- it's this function's main purpose, not an attribute it happens to have.

AddIndexToStringVolatileScalarUDF

findepi · 2024-10-22T10:48:14Z

datafusion/core/tests/user_defined/user_defined_scalar_functions.rs

@@ -483,6 +485,196 @@ async fn test_user_defined_functions_with_alias() -> Result<()> {
    Ok(())
 }

+/// Volatile UDF that should be append a different value to each row


Suggested change

/// Volatile UDF that should be append a different value to each row

/// Volatile UDF that should append a different value to each row

#[derive(Debug)]

findepi · 2024-10-22T10:48:43Z

datafusion/core/tests/user_defined/user_defined_scalar_functions.rs

+impl std::fmt::Debug for AddIndexToStringScalarUDF {
+    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
+        f.debug_struct("ScalarUDF")
+            .field("name", &self.name)
+            .field("signature", &self.signature)
+            .field("fun", &"<FUNC>")
+            .finish()
+    }
+}
+


Leverage #[derive(Debug)], to ensure all fields are part of debug.

Suggested change

impl std::fmt::Debug for AddIndexToStringScalarUDF {

fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {

f.debug_struct("ScalarUDF")

.field("name", &self.name)

.field("signature", &self.signature)

.field("fun", &"<FUNC>")

.finish()

}

}

findepi · 2024-10-22T10:49:37Z

datafusion/core/tests/user_defined/user_defined_scalar_functions.rs

+            _ => unimplemented!(),
+        };
+        Ok(ColumnarValue::Array(
+            Arc::new(StringArray::from(answer)) as ArrayRef


Suggested change

Arc::new(StringArray::from(answer)) as ArrayRef

Arc::new(StringArray::from(answer))

findepi · 2024-10-22T10:50:15Z

datafusion/core/tests/user_defined/user_defined_scalar_functions.rs

+            }
+            _ => unimplemented!(),
+        };
+        Ok(ColumnarValue::Array(


Is it OK for this function to return array also when it's invoked with ColumnarValue::Scalar only?

Yes, that's fine. If you return a ColumnarValue::Scalar, all rows will have the same result.

I was thinking about #12922

alamb · 2024-10-22T15:06:03Z

Thank you @agscpp and @findepi 🚀

Co-authored-by: Agaev Huseyn <[email protected]>

github-actions bot added logical-expr Logical plan and expressions physical-expr Physical Expressions core Core DataFusion crate labels Oct 18, 2024

agscpp force-pushed the fix_udf_function branch 2 times, most recently from ff4b90c to 4c8ebc4 Compare October 18, 2024 14:22

alamb reviewed Oct 18, 2024

View reviewed changes

agscpp force-pushed the fix_udf_function branch from 4c8ebc4 to 1e6ad19 Compare October 21, 2024 08:43

agscpp requested a review from alamb October 21, 2024 14:11

alamb approved these changes Oct 21, 2024

View reviewed changes

findepi approved these changes Oct 22, 2024

View reviewed changes

Fix functions with Volatility::Volatile and parameters

2c68fc9

agscpp force-pushed the fix_udf_function branch from 1e6ad19 to 2c68fc9 Compare October 22, 2024 11:25

agscpp requested a review from findepi October 22, 2024 11:28

alamb merged commit ef1365a into apache:main Oct 22, 2024
24 checks passed

alamb mentioned this pull request Oct 22, 2024

Deprecate ScalarUDFImpl::invoke #13064

Closed

agscpp added a commit to agscpp/datafusion that referenced this pull request Oct 22, 2024

Fix functions with Volatility::Volatile and parameters (apache#13001)

a3cbdea

Co-authored-by: Agaev Huseyn <[email protected]>

agscpp added a commit to tarantool/datafusion that referenced this pull request Oct 29, 2024

Fix functions with Volatility::Volatile and parameters (apache#13001)

2921595

Co-authored-by: Agaev Huseyn <[email protected]>

agscpp added a commit to tarantool/datafusion that referenced this pull request Oct 29, 2024

Fix functions with Volatility::Volatile and parameters (apache#13001)

01aa02d

Co-authored-by: Agaev Huseyn <[email protected]>

agscpp added a commit to tarantool/datafusion that referenced this pull request Oct 29, 2024

Fix functions with Volatility::Volatile and parameters (apache#13001)

28d1f3e

Co-authored-by: Agaev Huseyn <[email protected]>

This was referenced Oct 29, 2024

Deprecate invoke and invoke_no_args in favor of invoke_batch #13174

Merged

Migrate code to use ScalarUDF::invoke_batch and ScalarUDFImpl::invoke_batch #13238

Closed

askalt pushed a commit to tarantool/datafusion that referenced this pull request Nov 21, 2024

Fix functions with Volatility::Volatile and parameters (apache#13001)

ca4b62d

Co-authored-by: Agaev Huseyn <[email protected]>

0x501D pushed a commit to tarantool/datafusion that referenced this pull request Nov 21, 2024

Fix functions with Volatility::Volatile and parameters (apache#13001)

e64f889

Co-authored-by: Agaev Huseyn <[email protected]>

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Fix functions with Volatility::Volatile and parameters #13001

Fix functions with Volatility::Volatile and parameters #13001

agscpp commented Oct 18, 2024

alamb left a comment

alamb Oct 18, 2024

agscpp Oct 21, 2024

alamb Oct 18, 2024

agscpp Oct 21, 2024

agscpp Oct 21, 2024 •

edited

Loading

alamb Oct 18, 2024

alamb left a comment

alamb Oct 21, 2024

findepi Oct 22, 2024

alamb Oct 21, 2024

findepi Oct 22, 2024

findepi Oct 22, 2024

findepi Oct 22, 2024

agscpp Oct 22, 2024

findepi Oct 22, 2024

agscpp Oct 22, 2024

findepi Oct 22, 2024

findepi Oct 22, 2024

findepi Oct 22, 2024

findepi Oct 22, 2024

findepi Oct 22, 2024

findepi Oct 22, 2024

agscpp Oct 22, 2024 •

edited

Loading

findepi Oct 22, 2024

alamb commented Oct 22, 2024

-        not_impl_err!(
-            "Function {} does not implement invoke_batch but called",
-            self.name()
-        )
+  if _args.empty() {
+    self.invoke_no_args(number_rows)
+  } else {
+    self.invoke(args)
+  }

		/// The function should be used for signatures with [`datafusion_expr_common::signature::Volatility::Volatile`]
		/// and with arguments.

		/// Volatile UDF that should be append a different value to each row
		struct AddIndexToStringScalarUDF {

	/// Volatile UDF that should be append a different value to each row
	/// Volatile UDF that should append a different value to each row
	#[derive(Debug)]

	Arc::new(StringArray::from(answer)) as ArrayRef
	Arc::new(StringArray::from(answer))

Fix functions with Volatility::Volatile and parameters #13001

Fix functions with Volatility::Volatile and parameters #13001

Conversation

agscpp commented Oct 18, 2024

Which issue does this PR close?

Rationale for this change

alamb left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

agscpp Oct 21, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

alamb left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

agscpp Oct 22, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

alamb commented Oct 22, 2024

agscpp Oct 21, 2024 •

edited

Loading

agscpp Oct 22, 2024 •

edited

Loading