[EPIC] Unify Function Interface (remove `BuiltInScalarFunction`)

[alamb]

Is your feature request related to a problem or challenge?

This is based on the wonderful writeup from @2010YOUY01 in #7977

As previously discussed in #7110 #7752 there are a few challenges with how ScalarFunctions are handled, notably that there are two distinct implementations -- BuiltinScalarFunction and ScalarUDF

Problems with BuiltinScalarFunction

1. As more functions are added, the total footprint of DataFusion grows, even for those who don't need the specific functions. This also acts to limit the number of functions built into DataFusion
2. The desired semantics may be different for different users(e.g. many of the built in functions in DataFusion mirror postgres behavior, but some users wish to mimic spark behavior)
3. User defined functions are treated differently from built in functions in some ways (e.g. they can't have aliases)
4. Adding a new built in function requires modifications in multiple places which makes the barrier overly high.Built-in functions are implemented with Enum BuiltinScalarFunction, and function implementations like return_type() are large methods that match every enum variant.

Problems with ScalarUDF

• The current implementation of ScalarUDFs is a struct, and does not cover all the functionalities of existing built-in functions
• Defining a new ScalarUDF requires constructing a struct in an imperative way providing Arc function pointers (see examples/simple_udf.rs) for each part of the UDF, which is not familiar to Rust users where it is more common to see dyn Trait objects

Describe the solution you'd like

I propose moving DataFusion to only use ScalarUDFs and remove BuiltInScalarFunction. This will ensure:

1. ScalarUDFs have access to all the same functionality as "built in " functions.
2. No function specific code will escape the planning phase
3. DataFusion's core can remain focused, and external libraries of packages can be used to customize its use.

We will keep the existing ScalarUDF interface as much as possible, while also potentially providing an easier way to define them (ideally via a trait object).

Describe alternatives you've considered

#7977 describes introducing a new trait and unifying both ScalarUDF and BuiltInScalarFunction with this trait.

This approach also allows gradually migrating existing built-in functions to the new one, the old UDF interface create_udf() can keep unchanged.

However, I think it is a bigger change for users, and has the danger of making the overall complexity of DataFusion worse. As demonstrated in #8046 it is also feasible to allow new ScalarUDFs to be defined using a trait while retaining backwards compatibility for existing ScalarUDF implementations

Additional context

Proposed implementation steps:

• Prototype ScalarUDF interface changes (make the fields non pub): RFC: Make fields of ScalarUDF non pub #8039
• Prototype how registering external packages would look like (by making a prototype for some BuildInFunctions): RFC: Demonstrate what a function package might look like -- encoding expressions #8046
• Propose ScalarUDF API changes for real: Make fields of ScalarUDF , AggregateUDF and WindowUDF non pub #8079
• Minor: Remove redundant BuiltinScalarFunction::supports_zero_argument() #8059
• List additional feature gaps between built in functions and ScalarUDfs and close them
• Minor: Cleanup BuiltinScalarFunction's phys-expr creation #8114
• Unify Expr::AggregateFunction and Expr::AggregateUDF #8346
• Rename expr::window_function::WindowFunction to WindowFunctionDefintion for consistency #8347
• Implement monotonicity for ScalarUDF #8756
• Support Expr creation for ScalarUDF: Resolve function calls by name during planning #8157
• Add ScalarUDFs in missing function hints / suggested errors #9392
• Specialized / Pre-compiled / Prepared ScalarUDFs #8051
• Clean internal implementation of ScalarUDF to use ScalarUDFImpl (rather than the function pointers) #8712
• Minor: Cleanup BuiltinScalarFunction::return_type() #8088
• Implement aliasing for ScalarUDF: Implement Aliases for ScalarUDF #8348
• Implement trait based ScalarUDF: Implement trait based API for defining ScalarUDFs #8568
• [DISCUSS] organization of functions #9100
• Create a new datafusion-function crate with an initial set of functions as a model (see RFC: Demonstrate what a function package might look like -- encoding expressions #8046)
• Create tickets for extracting the remaining lists of packages into the datafusion_functions crate, file tickets to track them ([Epic] Port BuiltInFunctons to datafusion-functions-* crates #9285)
• [Epic] Port BuiltInFunctons to datafusion-functions-* crates #9285
• Add FunctionRegistry::register_udaf and FunctionRegistry::register_udwf #9074
• File follow on tickets for applying the same treatment to AggregateUDF [Epic] Unify AggregateFunction Interface (remove built in list of AggregateFunction s), improve the system #8708 and WindowUDF [Epic] Unify WindowFunction Interface (remove built in list of BuiltInWindowFunction s) #8709

[thinkharderdev]

Would like to add that supporting serialization of user-defined functions would be quite nice. The current approach to serialization is to basically just use a string and so if you have a scalar udf with some constant parameter you end up having to encode the constant as a literal expression which is less than ideal. Providing a mechanism (through PhysicalExtensionCodec I assume?) to serialize udfs would be very nice.

[alamb]

Would like to add that supporting serialization of user-defined functions would be quite nice.

I don't understand this question @thinkharderdev 🤔

The current approach to serialize a ScalarUDF as I understand it, is to send over the function name along with the arguments

https://github.com/apache/arrow-datafusion/blob/c2e768052c43e4bab6705ee76befc19de383c2cb/datafusion/proto/proto/datafusion.proto#L686-L689

And then the physical plan deserialization looks that function up by name https://docs.rs/datafusion-proto/latest/src/datafusion_proto/physical_plan/from_proto.rs.html#318-333

The only difference compared to BuiltInScalarFunction is that BuiltInScalarFunction uses a number (in an enum) which might be slightly faster

[thinkharderdev]

Would like to add that supporting serialization of user-defined functions would be quite nice.

I don't understand this question @thinkharderdev 🤔

The current approach to serialize a ScalarUDF as I understand it, is to send over the function name along with the arguments

https://github.com/apache/arrow-datafusion/blob/c2e768052c43e4bab6705ee76befc19de383c2cb/datafusion/proto/proto/datafusion.proto#L686-L689

And then the physical plan deserialization looks that function up by name https://docs.rs/datafusion-proto/latest/src/datafusion_proto/physical_plan/from_proto.rs.html#318-333

The only difference compared to BuiltInScalarFunction is that BuiltInScalarFunction uses a number (in an enum) which might be slightly faster

Sorry, what I mean is that it would be useful to be able to serialize constant parameters into the user-defined scalar function themselves rather than pass them in as expressions. So for instance if you had to create a UDF to do something with a regex that you have as a static constant. Currently the way to do that is pass it as a literal expression. But then you have to compile the regex again for every batch you process through the UDF. Ideally you could have something like:

struct MyRegexUdf {
  regex: Regex
}

impl ScalarFunction for MyRegexUdf {
  // use regex on each value somehow
}

The regex would only need to be compiled once during deserialization (or construction) instead of once for each batch.

[alamb]

Sorry, what I mean is that it would be useful to be able to serialize constant parameters into the user-defined scalar function themselves rather than pass them in as expressions

This makes total sense -- I filed #8051 to track the issue

[alamb]

Here is a PR that shows what DataFusion might look like after removing BuiltInScalarFunction and moving everything to ScalarUDF: #8046

[2010YOUY01]

According to the previous prototyping in #7978, we might need to do several cleanups towards this issue.

Following 2 is necessary for separating encode()/decode() and make API change (in #8046)

1. return_type() now do both input validation and generating output, these responsibilities have to be separated
2. resolve expr_fn using function string names as discussed in RFC: Demonstrate what a function package might look like -- encoding expressions #8046 (comment)

Functionalitiy gaps between BuiltinScalarFunction and ScalarUDF include

1. aliasing
2. monotonicity
3. Figure out the interface to let core pass information to function definitions (e.g. now() requires to be passed query start time from core) https://github.com/apache/arrow-datafusion/blob/308c35404939ed39e4e398054d6fc78c89817109/datafusion/physical-expr/src/functions.rs#L608

And other clean-up tasks might be helpful

1. Can they be moved into function implementations https://github.com/apache/arrow-datafusion/blob/308c35404939ed39e4e398054d6fc78c89817109/datafusion/physical-expr/src/functions.rs#L75-L204

[alamb]

Thank you for the investigation @2010YOUY01

monotonicity

This gap may be one of the gaps we are observing in IOx from a recent update from @mustafasrepo and @ozankabak in #8006 / https://github.com/apache/arrow-datafusion/issues/8043🤔

I will try and make it a priority to get a PR up to encapsuate the ScalarUDF fields so we can begin closing these gaps and I will update the list above

[alamb]

Figure out the interface to let core pass information to function definitions (e.g. now() requires to be passed query start time from core)

This is a similar usecase to what @thinkharderdev describes on #8051 -- I wrote up some ideas #8051 (comment)

[alamb]

Can they be moved into function implementations

@2010YOUY01 -- I totally agree this could be cleaned up. Moving it to specific implementations would be great. Could you make a PR to do so?