Fix functions with Volatility::Volatile and parameters by agscpp · Pull Request #13001 · apache/datafusion

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

        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

    ///
    /// [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

    ///
    /// [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

    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 Changes to the physical-expr crates 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

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]>

-        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))

Comments

Conversation

agscpp commented Oct 18, 2024

Which issue does this PR close?

Rationale for this change

Uh oh!

alamb left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

agscpp Oct 21, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

alamb left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

agscpp Oct 22, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

alamb commented Oct 22, 2024

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

agscpp Oct 21, 2024 •

edited

Loading

agscpp Oct 22, 2024 •

edited

Loading