feat: Add datafusion-spark crate

[andygrove]

Which issue does this PR close?

Closes #5600

Rationale for this change

See discussion in #5600

TL;DR Many projects want Spark-compatible expressions for use with DataFusion. We have some in Comet and there are also some in the Sail project. Let's move Comet's expressions into the core repo and collaborate there going forward.

What changes are included in this PR?

• Add copy of datafusion-comet-spark-expr crate with history
• Rename to datafusion-spark
• Add to workspace
• Pass CI checks

Are these changes tested?

Are there any user-facing changes?

[comphead]

I'm wondering how the user should choose what function to use. For instance to_timestamp function which may behave differently between non spark and spark env.

So the developer implements spark compliant to_timestamp function in the crate but when he is running the query
select to_timestamp() from t1 what exactly implementation will be picked up? Like we should introduce something like FUNCTION CATALOG or something similar to switch between implementations?

[andygrove]

I'm wondering how the user should choose what function to use. For instance to_timestamp function which may behave differently between non spark and spark env.

In the short term, the main goal is to allow downstream projects to add these functions to a DataFusion context, but users could manually configure their DataFusion context to choose which functions to use.

[viirya]

Hmm, it is good to add this crate. But I'm wondering that our other function crates like functions-nested should also follow Spark behavior, right? So it sounds like all these functions should be in datafusion-functions-spark?

[alamb]

🤔 you are right that it is kind of strange to have all spark functions in a single crate when we have the other functions in more fine grained crates

Maybe we should make one crate like datafusion_spark that holds all the different types of functions:

i    "datafusion/functions-spark",
+    "datafusion/spark",

And then over time we can add things like datafusion_spark_functions_nested in datafusion/spark/functions-nested 🤔

[andygrove]

Thanks @viirya. That is a good point. I renamed functions-spark to spark for now.

[alamb]

I'm wondering how the user should choose what function to use. For instance to_timestamp function which may behave differently between non spark and spark env.

So the developer implements spark compliant to_timestamp function in the crate but when he is running the query select to_timestamp() from t1 what exactly implementation will be picked up? Like we should introduce something like FUNCTION CATALOG or something similar to switch between implementations?

What I was imagining was that just like today DataFusion comes with some set if "included" functions (mostly following the postgres semantics). Nothing new is added to the core

For users that want to use the spark functions, I am hoping we have something like

use datafusion_functions_spark::install_spark;
...
// by default this SessionContext has all the same functions as it does today
let ctx = SessionContext::new();
// for users who want spark compatible functions, install the 
// spark compatibility functions, potentially overriding
// some/all "built in functions", rewrites, etc
install_spark(&mut ctx)?;
// running queries now use spark compatible syntax
ctx.sql(...)

[alamb]

not needed for this PR but I am intrested why there is a separate AVG implementation

[andygrove]

One difference is that this version uses an i64 for the count in the accumulator rather than a u64 (which makes sense because Java does not have unsigned ints).

[andygrove]

Of course, we should really have some documentation for these expressions explaining what the differences are.

[shehabgamin]

Subtle differences (e.g., i32 instead of i64, nullableinstead of non-nullable, microseconds instead of nanoseconds, etc.) between the DataFusion "default" implementation and the Spark implementation are going to be a common theme with many of these UDFs.

@andygrove @alamb This is where having something like #14296 (comment) would be useful. Otherwise we will always have to keep track of any changes made to the original DataFusion "default" implementation.

[alamb]

As I have mentioned previously I think this is a great idea.

In order to really move it into DataFusion though I think we should:

1. Have tests of this code in DataFusion (not via spark, but maybe via sqllogictests)
2. Have some way to register these functions so other users can use them

But in general the idea is 😍 to me

cc @Blizzara and @shehabgamin as you have mentioned interest in helping here

[alamb]

we would probably want to rename this eventually so it didn't refer to comet

[shehabgamin]

cc @Blizzara and @shehabgamin as you have mentioned interest in helping here

Will catch up on this thread tonight. So much happening so fast, exciting!

[shehabgamin]

Function Names

I'm wondering how the user should choose what function to use. For instance to_timestamp function which may behave differently between non spark and spark env.

In the short term, the main goal is to allow downstream projects to add these functions to a DataFusion context, but users could manually configure their DataFusion context to choose which functions to use.

What I was imagining was that just like today DataFusion comes with some set if "included" functions (mostly following the postgres semantics). Nothing new is added to the core

For users that want to use the spark functions, I am hoping we have something like

use datafusion_functions_spark::install_spark;
...
// by default this SessionContext has all the same functions as it does today
let ctx = SessionContext::new();
// for users who want spark compatible functions, install the 
// spark compatibility functions, potentially overriding
// some/all "built in functions", rewrites, etc
install_spark(&mut ctx)?;
// running queries now use spark compatible syntax
ctx.sql(...)

@comphead @andygrove @alamb It may be a good idea to prefix all function names with spark_ to avoid confusion, conflicts, or unknown behavior between functions that share the same name.

Subtle UDF Differences

See comment here: #14392 (comment)

Testing

In order to really move it into DataFusion though I think we should:

1. Have tests of this code in DataFusion (not via spark, but maybe via sqllogictests)

All functions that Sail ports over can be tested without the JVM or Spark Client. Because there is no JVM involvement when running a Sail server, it would be a relatively straightforward process to port function tests from Sail upstream.

1. Gold Data Tests -> SLT

Gold Data tests are SQL queries; functions get tested using Scalar input. We should be able to port over the vast majority of Gold Data tests for functions upstream to the SLTs.
The only caveat here is that Spark’s SQL semantics (Hive dialect) are different from DataFusion’s SQL semantics.
The implication of this is that some Gold Data tests can’t be ported upstream. However, I expect that the number of tests we can’t port over will be very small. The vast majority of Gold Data tests specifically for functions are very basic queries (e.g., SELECT ascii(2)).

2. Spark Functions Doc Tests -> SLT

These tests use the PySpark client (no JVM is required), and functions generally get tested using Array input (e.g., by calling a function with some DataFrame column as input). We should be able to translate almost all of these tests (if not all) to SQL.

Code Organization, Sail Contributions, etc.

It probably makes the most sense to merge this PR first before diving into these topics.

[jayzhan211]

It may be a good idea to prefix all function names with spark_ to avoid confusion, conflicts, or unknown behavior between functions that share the same name.

In optimizer, we rely on the name to do such optimization so if we rename it to name like 'spark_count' we might need to add the spark name to those optimize rules as well, which increase the maintainence cost. If we assume the datafusion native function and spark function is mutually exclusive (I guess we do so) then having consistent name for optimizer is preferred choice.

Have some way to register these functions so other users can use them

It will be cool if we can switch between spark and native function by the configuration (override functions in register if we switch the flag). It will be useful for testing in sqllogictest.

[shehabgamin]

In optimizer, we rely on the name to do such optimization so if we rename it to name like 'spark_count' we might need to add the spark name to those optimize rules as well, which increase the maintainence cost. If we assume the datafusion native function and spark function is mutually exclusive (I guess we do so) then having consistent name for optimizer is preferred choice.

I'm glad you brought this up, @jayzhan211.

Some Spark functions behave identically to DataFusion functions but have different names. For example:

• Spark’s startswith(str, substr) corresponds to DataFusion’s expr_fn::starts_with(str, substr)

There are also cases where functions take input arguments in a different order. For example:

• Spark’s position(substr, str) corresponds to DataFusion’s expr_fn::strpos(str, substr)

[jayzhan211]

In optimizer, we rely on the name to do such optimization so if we rename it to name like 'spark_count' we might need to add the spark name to those optimize rules as well, which increase the maintainence cost. If we assume the datafusion native function and spark function is mutually exclusive (I guess we do so) then having consistent name for optimizer is preferred choice.

I'm glad you brought this up, @jayzhan211.

Some Spark functions behave identically to DataFusion functions but have different names. For example:

• Spark’s startswith(str, substr) corresponds to DataFusion’s expr_fn::starts_with(str, substr)

There are also cases where functions take input arguments in a different order. For example:

• Spark’s position(substr, str) corresponds to DataFusion’s expr_fn::strpos(str, substr)

If the behavior is the same we don't have any reason to copy one to spark crate, adding alias to the function is enough.

If the diff between the native and spark one is minor we can also add another flag to switch code logic instead of maintaining a copy in another crate so that the maintenance cost could be minimized.

[shehabgamin]

If the behavior is the same we don't have any reason to copy one to spark crate, adding alias to the function is enough.

@jayzhan211 I’ll need to verify this, but I recall encountering a case in DataFusion where some two function names, A and B, are mapped inversely in Spark. Specifically, some Spark's B corresponds to some DataFusion's A, and some Spark's A corresponds to some DataFusion's B.

[andygrove]

@comphead @andygrove @alamb It may be a good idea to prefix all function names with spark_ to avoid confusion, conflicts, or unknown behavior between functions that share the same name.

For Comet, we need the function names to match Spark.

[andygrove]

For Comet, we need the function names to match Spark.

Actually, this isn't true. In Comet, we just need the Scala wrapper classes to have the same function names as Spark.

[comphead]

That is exactly what I was mentioning #14392 (comment) not sure how to separate out functions with the same names. I think this crate and PR makes a lot of sense until the user wants a mixed implementation, another thing is the code is duplicated, so changes in base function may not be reflected in spark variant.

@andygrove @alamb maybe we can do a conditional compilation on function level instead of separate crate?
Sort of introduce a feature spark and have different implementations

#[cfg(spark)]
fn to_timestamp() {}

#![cfg(spark)]
fn to_timestamp() {}

[andygrove]

I'm going to move this to draft for now. Let's keep the conversation going, though.

I plan on working with the community to start adding tests to the datafusion-comet-spark-expr crate that don't depend on Spark, and to add some examples, as those are two large areas of valid concern.