Current state: [One of "Under Discussion", "Accepted", "Rejected"]
Discussion thread:
JIRA or Github Issue:
Released: <Doris Version>
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.>
Support for Java UDF is motivated by the following points:
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.
We should use Doris Create-Function statement before. To ensure proper and more general semantics, we should update some properties.
A simple example for creating a JAVA_UDF is:
To ensure correctness, we should do some analysis for user jar including if file exist, if class name is legal, etc.
First of all, 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..
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:
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:
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 a initial buffer size 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. So this process is illustrated as figure 2:
figure 2[1]
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.
figure 3
Notice that, in step 1, we use an additional operation to convert column-cased input to row-based.
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.
[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 2015.