Daft Functions Roadmap

[kevinzwang]

Recently, we created a new daft.functions module which we hope to expand upon. The work here will serve two purposes:

1. Move Daft away from the namespaced expression syntax (e.g. col("a").str.capitalize()) which users have found confusing
2. Allow functions to be used with scalar values in addition to expressions, similar to our new UDFs (see New UDFs Roadmap #4820)

For each relevant method on daft.Expression or its namespaces, we will do the following:

• If in a namespace, flatten and rename it so that it sits as a direct method of Expression. Keep the original but add a deprecation warning once the move is complete. We'll remove it in v0.6
• Create a corresponding function in daft.functions with the same name
  • If applicable, make it also work for scalar Python values (e.g. explode is only applicable to expressions)
  • The idea is to allow both composition and chaining, but default to composition. For some functions, such as if_else, there may not need to be an expression method variant.
• Move docs to the function, add a link to the function in the expression method docstring
• Use function in tests and docs instead of expression method
• In addition to the Python side, there is some cleanup on the Rust side we should do here too. In particular, some expressions live under the main Expr enum or as a FunctionExpr but should really be implemented as a ScalarFunction

Tasks

• Move Expression methods
• Establish structure of docs page
  • docs: categorize functions and make each function its own page #4838
• Move str namespace
• Move dt namespace
• Move embedding namespace
• Move float namespace
• Move url namespace
• Move list namespace
• Move struct namespace
• Move map namespace
• Move image namespace
• Move partitioning namespace
• Move json namespace
• Move binary namespace
• Turn on deprecation warnings and change docstrings for Expression methods to point to function

[plotor]

Hi @kevinzwang, can it be understood that there will be no concept of Expression in the future, and all will be uniformly called Function, but distinguished between system built-in functions (referred to as BIF) and User-defined functions (referred to as UDF)?

Then what's the difference between BIF and UDF? What I can think of is:

1. At the implementation level: UDFs are usually implemented in Python, while most BIFs are implemented in Rust, and they can also be implemented in Python too;
2. At the usage level: are UDF and BIF no longer differentiated for users? For example:

• New UDF Example

@daft.func
def my_udf(x: int) -> str:  # return dtype is inferred from type hint
    return f"{input}"

df.with_column("y", my_udf(col("x")))

• New BIF Example

Take the built-in json.query expression as an example from Daft docs, the usage based on the expression is as follows:

df = daft.from_pydict({
    "json": [
        '{"a": 1, "b": 2}',
        '{"a": 3, "b": 4}',
    ],
})
df = df.with_column("a", df["json"].json.query(".a"))

If we abandon the concept of Expression and switch to BIF, will it evolve into the following usage style? (It is assumed that json_query is a system built-in function and is functionally equivalent to the json.query Expression):

df = df.with_column("a", json_query(df["json"], ".a"))

This example is mainly to show that for users, df.with_column("y", my_udf(col("x"))) and df.with_column("a", json_query(df["json"], ".a")) are the same in usage.

[kevinzwang]

Hi @plotor, thanks for the questions! Yep, your example is exactly how we are envisioning Daft functions as well as UDFs to work.

can it be understood that there will be no concept of Expression in the future

Expressions will still exist and be a critical part of Daft. You can think of Daft functions as functions that operate on expressions. Moreover, most functions will also have an equivalent Expression method, solely to allow for method chaining. E.g. we will support both the following:

df["json"].json_query(".a")
json_query(df["json"], ".a")

The difference between the existing Expression methods and functions though, is that Daft functions can either produce expressions or Python values, depending on its inputs.

For example, if you called daft.functions.str_split(col("a"), " "), you will get an expression, but if you called daft.functions.str_split("foo bar", " "), you'd get a list ["foo", "bar"]. Essentially, Daft functions can be used lazily (via expressions) or eagerly (via Python objects).

This enables pretty cool things, such as the ability to use Daft functions within UDFs. For example, if you had a UDF that conditionally downloaded a file, it would be trivial to do with the url_download function.

Then what's the difference between BIF and UDF?

In terms of usage, there will be no difference. The only difference is how they are defined. Built-in functions are defined in the Daft library, whereas UDFs are defined by a user.

Some clarifying notes about this:

• You are exactly right that Daft may even have some built-in functions that are actually implemented under the hood as a Python UDF. Then, in the future, we may port it to Rust for better performance. But for a user, they will see no difference other than faster execution.
• We may expose ways to write UDFs in other languages in the future. But to a person building a Daft query, they will also interact with it just like a built-in function or a Python UDF.

[plotor]

@kevinzwang That is to say, BIF and UDF will be converted to Expression for execution, so that operators like with_column can accept both Expression and Function as parameters. However, I still have some small questions:

1. Is it Daft's internal work to convert Function to Expression? In this way, users don't need to pay attention to the implementation and details of UDF to Expression conversion when defining UDF.

2. I have previously planned to design and implement an Expression Plugin function, similar to Polars Expression Plugins, which is aimed at some functions that are not suitable for implementation in Daft, but want to be written in Rust and provided to users in Expression usage style. If New UDF supports writing in Rust, then can I understand that the Expression Plugin function is no longer needed, because UDF written in Rust already supports automatic conversion to Expression for use?

[kevinzwang]

1. I'm not 100% sure what you mean with that question. Functions/UDFs are not expressions themselves, but if you call them with expressions, they will produce expressions. As with my example above, str_split itself is not an expression, but str_split(col("a"), " ") returns an expression, which denotes the split of column "a".
2. Instead of something like Polars expression plugins, which is Rust-specific, what we could build instead is some sort of Arrow-backed API for writing UDFs in any arbitrary language. This can be done by either exposing a C API with FFI bindings for other languages, or some sort of gRPC framework/Arrow Flight. We haven't fleshed it out yet but it's something people have increasingly been asking about. Could you expand on your potential use case, and would that kind of interface work for you?

[plotor]

The reason why I want to plan the function of Expression Plugin is that before the release of New UDF, the difference between Old UDF and Expression is still very obvious:

1. UDF is user-defined and implemented in Python, while Expression is built-in and provided by Daft, mainly implemented in Rust.
2. UDF is called in a form similar to df.with_column("a", json_query(df["json"], ".a")), while Expression is called in a form of df.with_column("a", df["json"].json.query(".a")).

Considering the execution performance of UDF and the expectation of providing the same usage as Expression, as a toB cloud service platform, we hope that some of our UDFs can be implemented in Rust, and these UDFs can be used by users like Expression, just like they are natively provided by Daft, but these UDFs may not be directly contributed to the Daft community due to legal, non-universal, or binding to the platform's internal interfaces. In this context, I plan to implement an Expression Plugin feature to connect the Rust version of UDF with the Daft engine itself.

[kevinzwang]

One potential way to do this without any need for a plugins system in Daft itself is to write your Rust UDFs using Arrow (via arrow-rs), and then expose those Rust UDFs to Python using pyo3 and pyo3-arrow. Then, you only need to build a thin wrapper around the batch UDF:

@daft.func.batch
def my_udf(input: daft.Series):
    return my_udf_rust_impl(input.to_pyarrow())

This does still need to go through Python for execution, but the Rust function call releases the GIL, so with a good batch size, the Python code should not be a noticeable performance bottleneck. Additionally, the use of Arrow makes all of this zero copy across languages.

[rchowell]

Do we think calling like normal functions would be confusing for customers because the DataFrame API supports passing ColumnInputType in many places?

df.select("a").with_column("b", do_work("a"))  # !!! ERROR !!! do_work will be evaluated on string literal

Here we have "a" as a column reference in one context, but a string literal in another context.

[plotor]

Hi @kevinzwang @rchowell, Have we considered supporting switching Conda env when executing udf?

[kevinzwang]

Hi @plotor, isolated dependencies for UDFs is something we have thought about in the past but is not something we have fleshed out as a concept yet. What would be your use case for switching conda environments? Is it possible to just install the dependencies on the entire cluster?

[plotor]

@kevinzwang The main scenario is version conflicts between third-party dependencies between different UDFs. The current workaround is to write intermediate results to disk and then run a separate new job to handle them. I've considered two solutions:

1. Configure multiple Conda envs for the entire cluster, and flexibly switch between them based on the UDF configuration when executing the UDFs;

2. Set up multiple Ray workgroups, install a different Conda env in each workgroup, and use Ray custom resources to schedule different UDFs to run on different workgroup nodes.

Overall, solution 1 offers a better user experience and does not rely on Ray.

[kevinzwang]

The main scenario is version conflicts between third-party dependencies between different UDFs.
Makes sense. We don't have a plan for this yet but if you'd like to write up a draft proposal/design I would be happy to review it!

[plotor]

The main scenario is version conflicts between third-party dependencies between different UDFs.
Makes sense. We don't have a plan for this yet but if you'd like to write up a draft proposal/design I would be happy to review it!

OK, I've created an issue to track it at #5104. I'll research it and come up with a design.