Support Serde for ScalarUDF in Physical Expressions #9436

yyy1000 · 2024-03-03T03:32:30Z

Which issue does this PR close?

Closes #9428.

Rationale for this change

See the issue.

What changes are included in this PR?

To do

Are these changes tested?

Are there any user-facing changes?

yyy1000

Hi @thinkharderdev
I began this PR for physical-expr and I have a question.
Would like to see your thought!

yyy1000 · 2024-03-03T03:35:44Z

datafusion/proto/src/physical_plan/to_proto.rs

-        } else if let Some(expr) = expr.downcast_ref::<LikeExpr>() {
+        } else {
+            let mut buf = Vec::new();
+            // let _ = codec.try_encode_udf(, &mut buf);


I have a question, how can I get a ScalarUDF from a ScalarFunctionExpr?
Given its current structure, I would think change the fields in ScalarFunctionExpr may be a method.

I also tried using create_udf, but it seems can not work.

I'm assuming this is what you mean in your below comment but I think we need to change

pub struct ScalarFunctionExpr { fun: ScalarFunctionImplementation,

to

pub struct ScalarFunctionExpr { fun: ScalarFunctionDefinition,

@alamb This would technically be a breaking change since ScalarFunctionExpr has a public method

pub fn fun(&self) -> &ScalarFunctionImplementation { &self.fun }

We could technically do this is in a backwards compatible way I think by creating a corresponding ScalarFunctionImplementation from the ScalarFunctionImplementation or we could just change the public API. Seems to me like this would probably not be a very disruptive breaking change since ScalarFunctionImplementation is just an opaque function pointer so unlikely this is being used in application code in a way that wouldn't be easily adaptable if it started returning &ScalarFunctionDefinition instead

Thanks for your review! @thinkharderdev
Yeah, the change is what I thought. And I also think some other fields in ScalarFunctionExpr can be removed like name and return_type.

Yeah, I think create a ScalarFunctionImplementation from ScalarFunctionDefinition is a method. Given the BuiltinScalarFunction will be removed and we can get the inner of a ScalarUDF which is a ScalarUDFImpl, and we can do something like.

let captured = inner.clone(); Arc::new(move |args| captured.invoke(args))

I'd like to see the opinions from you and @alamb :)

yyy1000 · 2024-03-03T03:41:11Z

datafusion/physical-expr/src/scalar_function.rs

 };

 /// Physical expression of a scalar function
 pub struct ScalarFunctionExpr {
-    fun: ScalarFunctionImplementation,
+    fun: ScalarFunctionDefinition,


I think the fun here can be a ScalarFunctionDefinition to get the ScalarUDF from ScalarFunctionExpr?
Also I searched the code base and found ScalarFunctionImplementation is only used for:

BuiltinScalarFunction

method like create_udf which I think will be deprecated in the future given the new ScalarUDFImpl way.
So I think ScalarFunctionImplementation will also be deleted in a day. 🤔 WDYT?

yeah, this should be ScalarFunctionDefinition here. I don't think create_udf will be deprecated but we have ScalarUdfLegacyWrapper to wrap ScalarFunctionImplementation and create a ScalarUDFImpl out of it. So the new ScalarFunctionDefinition would include anything that is represented as a ScalarFunctionImplementation

yyy1000 · 2024-03-11T04:53:42Z

I think I need some feedback on #9436 (comment) and #9436 (comment) to continue this PR. 🤔

thinkharderdev · 2024-03-11T09:53:16Z

I think I need some feedback on #9436 (comment) and #9436 (comment) to continue this PR. 🤔

I think it is fine to replace ScalarFunctionImplementation with ScalarUDF and end up with something like:

pub struct ScalarFunctionExpr {
    fun: ScalarUDF,
    args: Vec<Arc<dyn PhysicalExpr>>,
    // Whether this function can be invoked with zero arguments
    supports_zero_argument: bool,
}

impl ScalarFunctionExpr {
  // Change this to return &ScalarUDF instead of &ScalarFunctionImplementation
  fn fun(&self) -> &ScalarUDF {
    &self.fun
  ]
}

yyy1000 · 2024-03-11T20:51:02Z

I think it is fine to replace ScalarFunctionImplementation with ScalarUDF and end up with something like:

pub struct ScalarFunctionExpr {
    fun: ScalarUDF,
    args: Vec<Arc<dyn PhysicalExpr>>,
    // Whether this function can be invoked with zero arguments
    supports_zero_argument: bool,
}

impl ScalarFunctionExpr {
  // Change this to return &ScalarUDF instead of &ScalarFunctionImplementation
  fn fun(&self) -> &ScalarUDF {
    &self.fun
  ]
}

A question is, currently there're still BuiltInFunctions, should ScalarUDF be ScalarFunctionDefinition?
Later after porting is finished we can use ScalarUDF.

thinkharderdev · 2024-03-11T21:09:18Z

I think it is fine to replace ScalarFunctionImplementation with ScalarUDF and end up with something like:
pub struct ScalarFunctionExpr {
    fun: ScalarUDF,
    args: Vec<Arc<dyn PhysicalExpr>>,
    // Whether this function can be invoked with zero arguments
    supports_zero_argument: bool,
}

impl ScalarFunctionExpr {
  // Change this to return &ScalarUDF instead of &ScalarFunctionImplementation
  fn fun(&self) -> &ScalarUDF {
    &self.fun
  ]
}
A question is, currently there're still BuiltInFunctions, should ScalarUDF be ScalarFunctionDefinition? Later after porting is finished we can use ScalarUDF.

Yeah I think that's correct

…afusion into serde-phsyical

yyy1000 · 2024-03-12T18:06:21Z

I'm waiting on this #9537 to be merged.
This can reduce some effort caused by ExecutionProps when create a physical scalar function from BuiltIn

yyy1000

Hi @thinkharderdev
I finished the PR for Serde in PhysicalExpr, though a little delay
Could you review it when you are available?
Thanks!

yyy1000 · 2024-03-13T21:05:01Z

datafusion/core/src/physical_optimizer/projection_pushdown.rs

-                Arc::new(|_: &[ColumnarValue]| unimplemented!("not implemented")),
+                ScalarFunctionDefinition::UDF(Arc::new(ScalarUDF::new_from_impl(
+                    DummyUDF::new(),
+                ))),


Since here ScalarFunctionExpr needs a ScalarFunctionDefinition, I created one for each. Don't know whether there's a better way to do it.

yyy1000 · 2024-03-13T21:06:22Z

datafusion/physical-expr/src/functions.rs

@@ -57,7 +58,7 @@ pub fn create_physical_expr(
    fun: &BuiltinScalarFunction,
    input_phy_exprs: &[Arc<dyn PhysicalExpr>],
    input_schema: &Schema,
-    execution_props: &ExecutionProps,
+    _execution_props: &ExecutionProps,


Here the ExecutionProps param can be moved, but I'd like to do it in a follow-up PR because doing all in a PR may make me less easy to inspect error.

weird, seems like it never needed this param

Yes, it doesn't need this param now, but before it needs this for BuiltInFunction to create physical function for some Date function like to_date. Since these functions has been ported, it will not need this param anymore.

yyy1000 · 2024-03-13T21:07:17Z

datafusion/physical-expr/src/functions.rs

    let monotonicity = fun.monotonicity();

+    let fun_def = ScalarFunctionDefinition::BuiltIn(*fun);


Here I just pass the fun_def, will call create_physical_fun when execute the PhysicalExpr

yyy1000 · 2024-03-13T21:08:32Z

datafusion/physical-expr/src/scalar_function.rs

@@ -171,8 +171,17 @@ impl PhysicalExpr for ScalarFunctionExpr {
                .collect::<Result<Vec<_>>>()?,
        };

+        let fun_implementation = match self.fun {
+            ScalarFunctionDefinition::BuiltIn(ref fun) => create_physical_fun(fun)?,


Here it will call create_physical_fun to get the FunctionImplementation

yyy1000 · 2024-03-13T21:09:15Z

datafusion/proto/src/physical_plan/from_proto.rs

-                let expr = parse_physical_expr(expr.as_ref(), registry, input_schema)?;
+                let codec = DefaultPhysicalExtensionCodec {};
+                let expr =
+                    parse_physical_expr(expr.as_ref(), registry, input_schema, &codec)?;


I create DefaultPhysicalExtensionCodec for cases that will not need ExtensionCodec

yyy1000 · 2024-03-13T21:09:50Z

datafusion/proto/src/physical_plan/from_proto.rs

+            let udf = match &e.fun_definition {
+                Some(buf) => codec.try_decode_udf(&e.name, buf)?,
+                None => registry.udf(e.name.as_str())?,
+            };


Here is the decode logic

yyy1000 · 2024-03-13T21:10:46Z

datafusion/proto/src/physical_plan/to_proto.rs

+    value: Arc<dyn PhysicalExpr>,
+    codec: &dyn PhysicalExtensionCodec,
+) -> Result<protobuf::PhysicalExprNode, DataFusionError> {
+    let expr = value.as_any();


Applied the method from LogicalExpr

yyy1000 · 2024-03-13T21:11:50Z

datafusion/proto/tests/cases/roundtrip_physical_plan.rs

@@ -665,6 +670,133 @@ fn roundtrip_scalar_udf() -> Result<()> {
    roundtrip_test_with_context(Arc::new(project), ctx)
 }

+#[test]


Here is the test case, mainly copied from LogicalExpr Serde test

yyy1000 · 2024-03-13T21:25:07Z

I believe that CI fails because of some system glitch, so this PR is still ready for review.

thinkharderdev

nice work! I had a few nitpicks but otherwise I think this is ready

thinkharderdev · 2024-03-19T08:29:51Z

datafusion/physical-expr/src/functions.rs

@@ -57,7 +58,7 @@ pub fn create_physical_expr(
    fun: &BuiltinScalarFunction,
    input_phy_exprs: &[Arc<dyn PhysicalExpr>],
    input_schema: &Schema,
-    execution_props: &ExecutionProps,
+    _execution_props: &ExecutionProps,


weird, seems like it never needed this param

datafusion/physical-expr/src/scalar_function.rs

datafusion/proto/src/physical_plan/to_proto.rs

datafusion/proto/tests/cases/roundtrip_physical_plan.rs

thinkharderdev · 2024-03-19T09:04:01Z

I believe that CI fails because of some system glitch, so this PR is still ready for review.

Yep, I re-ran the job and it worked

Co-authored-by: Dan Harris <1327726+thinkharderdev@users.noreply.github.com>

yyy1000 · 2024-03-19T15:34:35Z

Thanks @thinkharderdev for your review!
I addressed the comment and I think it's good to go now :)

thinkharderdev · 2024-03-19T17:00:41Z

Thanks @thinkharderdev for your review! I addressed the comment and I think it's good to go now :)

Thank you for your work on this!

initial try

9ab42e2

github-actions bot added the physical-expr Physical Expressions label Mar 3, 2024

yyy1000 commented Mar 3, 2024

View reviewed changes

yyy1000 marked this pull request as draft March 3, 2024 03:42

revert

8d7c866

yyy1000 marked this pull request as ready for review March 11, 2024 04:52

yyy1000 added 2 commits March 12, 2024 11:32

Merge branch 'main' into serde-phsyical

ee9c36f

Merge branch 'serde-phsyical' of https://github.com/yyy1000/arrow-dat…

6ecfb11

…afusion into serde-phsyical

yyy1000 marked this pull request as draft March 12, 2024 17:33

yyy1000 added 3 commits March 12, 2024 13:12

stage commit

6b154e0

Merge branch 'main' into serde-phsyical

ecde242

use ScalarFunctionDefinition to rewrite PhysicalExpr proto

b33656a

github-actions bot added the core Core DataFusion crate label Mar 13, 2024

yyy1000 added 4 commits March 13, 2024 14:55

cargo fmt

1738529

feat : add test

93d9750

fix bug

376905b

fix wrong delete code when resolve conflict

d8d1eb2

yyy1000 marked this pull request as ready for review March 13, 2024 21:03

yyy1000 commented Mar 13, 2024

View reviewed changes

yyy1000 requested a review from thinkharderdev March 19, 2024 00:35

thinkharderdev approved these changes Mar 19, 2024

View reviewed changes

yyy1000 and others added 2 commits March 19, 2024 10:03

Update datafusion/proto/src/physical_plan/to_proto.rs

623e56c

Co-authored-by: Dan Harris <1327726+thinkharderdev@users.noreply.github.com>

Update datafusion/proto/tests/cases/roundtrip_physical_plan.rs

1dfce4a

Co-authored-by: Dan Harris <1327726+thinkharderdev@users.noreply.github.com>

address the comment

767b53d

thinkharderdev merged commit 0974759 into apache:main Mar 19, 2024
23 checks passed

yyy1000 deleted the serde-phsyical branch March 19, 2024 18:03

yyy1000 mentioned this pull request Mar 19, 2024

Refine Serde for Scalar Function #9699

Closed

joroKr21 mentioned this pull request Apr 14, 2024

Use PhysicalExtensionCodec consistently #10075

Merged

joroKr21 mentioned this pull request Jul 10, 2024

Add customizable equality and hash functions to UDFs #11392

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Support Serde for ScalarUDF in Physical Expressions #9436

Support Serde for ScalarUDF in Physical Expressions #9436

yyy1000 commented Mar 3, 2024

yyy1000 left a comment

yyy1000 Mar 3, 2024

yyy1000 Mar 3, 2024

thinkharderdev Mar 3, 2024

yyy1000 Mar 3, 2024

yyy1000 Mar 3, 2024

thinkharderdev Mar 3, 2024

yyy1000 commented Mar 11, 2024

thinkharderdev commented Mar 11, 2024

yyy1000 commented Mar 11, 2024

thinkharderdev commented Mar 11, 2024

yyy1000 commented Mar 12, 2024

yyy1000 left a comment

yyy1000 Mar 13, 2024

yyy1000 Mar 13, 2024

thinkharderdev Mar 19, 2024

yyy1000 Mar 19, 2024

yyy1000 Mar 13, 2024

yyy1000 Mar 13, 2024

yyy1000 Mar 13, 2024

yyy1000 Mar 13, 2024

yyy1000 Mar 13, 2024

yyy1000 Mar 13, 2024

yyy1000 commented Mar 13, 2024

thinkharderdev left a comment

thinkharderdev Mar 19, 2024

thinkharderdev commented Mar 19, 2024

yyy1000 commented Mar 19, 2024

thinkharderdev commented Mar 19, 2024

		let monotonicity = fun.monotonicity();

		let fun_def = ScalarFunctionDefinition::BuiltIn(*fun);

Support Serde for ScalarUDF in Physical Expressions #9436

Support Serde for ScalarUDF in Physical Expressions #9436

Conversation

yyy1000 commented Mar 3, 2024

Which issue does this PR close?

Rationale for this change

What changes are included in this PR?

Are these changes tested?

Are there any user-facing changes?

yyy1000 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

yyy1000 commented Mar 11, 2024

thinkharderdev commented Mar 11, 2024

yyy1000 commented Mar 11, 2024

thinkharderdev commented Mar 11, 2024

yyy1000 commented Mar 12, 2024

yyy1000 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

yyy1000 commented Mar 13, 2024

thinkharderdev left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

thinkharderdev commented Mar 19, 2024

yyy1000 commented Mar 19, 2024

thinkharderdev commented Mar 19, 2024