Status

Current state:  [One of "Under Discussion", "Accepted", "Rejected"] Released

Discussion thread:  https://lists.apache.org/thread/31jrw7tmd40031nqp7l50jzfdhnqxj0g

JIRA or Github Issue: 

Released:  <Doris Version>1.2.0

Google Doc: <If the design in question is unclear or needs to be discussed and reviewed, a Google Doc can be used first to facilitate comments from others.>

Motivation

Describe the problems you are trying to solve.

Related Research

some research related to the function, such as the advantages and disadvantages of the design, related considerations, etc.

Detailed Design

the detailed design of the function.

Scheduling

Support for Java UDF is motivated by the following points:

1. the Java language is widely used, rich support for related library functions, and a low threshold for development and debugging.
2. Some big data ecologies such as Hive, Spark, etc. already exist a large number of ready-made UDF, we need to be compatible with these udf to reduce the migration costs of users.
3. If there is bug in UDF, it will not cause BE crash

Related Research

Calling Java UDF is common use case for native code. In this process, calling JNI is inevitable. As we know, improper JNI calls will causes serious performance issues. So a more effecient way is using strided execution which means call JNI for multiple rows[1]. In vectorized engine, it means we only need call JNI once for one or more input columns.
Besides, Apache Impala also provide a way to implement Java UDFs[2]. Apache Impala use a row-based query engine so it is more natural when dealing with scalar functions. In Doris vectorized query engine, Java UDF is more challenging.

Detailed Design

1 How to Create Java UDF

We should use Doris Create-Function statement before. To ensure proper and more general  semantics, we should update some properties.

1. Rename `object_file` to `file`. `file` is more general to present a user file. To ensure compatibility, we also retain `object_file` but mark as `depracated`
2. `symbol` for JAVA_UDF indicates class in UDF jar.
3. Add JAVA_UDF type.

A simple example for creating a JAVA_UDF is:

Image Added

To ensure correctness, we should do some analysis for user jar including if file exist, if class name is legal, etc.

2 Execution on Vectorized Engine

First of all, a new Java function call will be executed in Doris query engine and BE will create or reuse a JVM to call the real Java UDF. To isolate different UDF instances, we use different class loader to load UDF.

Secondly, it’s a better way to implement JNI call using strided execution than row-by-row execution. As mentioned in paper [1], this leads to a much better performance since the JNI overhead is amortized by all rows in input columns.

Notably, user must follow certain rules when creating UDFs. For example, UDF class must have evaluate  method and it's must be public and non-static. These rules ensure we can invoke UDF correctly.

2.1 Fixed-length output

To use strided execution mode, a basic idea is passing addresses which point to input buffer and output buffer directly. This can help us to avoid unnecessary data copies. Input buffer and output buffer are both JVM off-heap memory and fortunately, Java provides API to manipulate off-heap memory for us.

So overall execution mode is illustrated in figure 1:

Image Added

figure 1

Step 1, allocate output buffer for UDF.

Step 2, pass addresses point to input buffer and output buffer to FE.

Step 3, execute a for-loop over all columns and pass i-th elements to UDF and finally output i-th result into result buffer. Pseudo code for this process is below:

Image Added

2.2 Variable-length output

For fixed-length type input and output, this is a standardized process because each buffer size is fixed. But for variable-length output type, above steps are no longer applicable. For variable-length output, we always allocate an initial buffer in Step 1, and jump out of Step 3 when size of results is bigger than initial buffer allocated in Step 1. When it happens, we repeat Step 1 ~3 again to allocate a new buffer and continue to execute UDF for remain rows. To do this, we should maintain some states to ensure correctness. So this process is illustrated as figure 2:

Image Added

figure 2^[1]

2.3 Convert input columns to rows

In some cases, above process works well. Unfortunately, it is not a perfect choice for each case. Assume that each input column can fill the cache, when we access each row in each column in Step 3, every access will incur a cache miss. So in the worst case, Step 3 will incur numRows * numColumns cache misses which is a performance disaster.

So we should copy data properly to improve data locality as illustrated in figure 3.

Image Added

figure 3

Notice that, in step 1, we use an additional operation to convert column-cased input to row-based.

Although this method can avoid cache missed as much as possible, the most critical issue is when we should use a row-based input to call a UDF. This is a trade-off between data copies and cache misses. This strategy is still unclear and we need to discuss later.

Scheduling

I have a preliminary plan to support Java UDF in Doris.

Phase I, I'll complete Create-Function-Statement for Java UDF and column-based fixed-length UDF (described in detailed design - 2 - 2.1).

Phase II, I'll complete all type column-based UDF and row-based UDF (described in detailed design - 2 - 2.2 && 2.3).

Phase III, after completing Java UDF, I'll complete jobs about Java UDAF and UDTF.

References

[1] Viktor Rosenfeld, René Müller, Pinar Tözün, etc. Processing Java UDFs in a C++ environment. SoCC 2017: 419-431.

[2] Marcel Kornacker, Alexander Behm, Victor Bittorf, etc. Impala: A Modern, Open-Source SQL Engine for Hadoop. CIDR 2015specific implementation steps and approximate scheduling.