Java API for Velox plan

[hn5092]

Description

Generally speaking, to invoke Velox from Java now, one needs to construct from the Substrait side, creating a Substrait execution plan. The whole usage and learning cost is very high, requiring a deep understanding of the entire SQL and also a profound knowledge of Substrait. This necessitates mastering a lot of things, and for developers, it is very inconvenient. Complexity implies that the cost of bug maintenance is huge. Here, I propose the idea of building a JAVA SDK.

Goals:

1. Scan a big list

case class Data(A: Int, b: String)
velox.values(Array[Data]).agg("b", "count(a)").collect

2. Scan a formatted file

velox.scan.parquet("filePath", schema, conf).agg("b", "count(a)").collect

3. Generate a stage execution plan

velox.scan.parquet("filePath", schema, conf).select("a", "b").planJson

What needs to be done:

1. JNI framework:

C++:
a. Native class, implementing this object can easily interface with a Java class.
b. Framework access to Java objects is also simple.
c. High performance.

Java:
a. Native class, an abstract class of the entire native object, inside which some checks can be done. Implementing this method of objects will have some default behaviors to prevent crashes like this.

2. Function resolution

3. Transfer of intermediate results

4. Velox Vector JAVA API

[hn5092]

Our engine, Starry DB, has already implemented a similar set of APIs, from which we have reaped huge benefits. Two of us completed the development and launch of this Spark native engine in half a year. If possible, we would like to share all of this with everyone.

[majetideepak]

@hn5092 do you have any documentation of reference for these APIs?

[zhztheplayer]

This could be very helpful for JVM frontends.

case class Data(A: Int, b: String)

Curious did you mean Java + Scala SDK? Since it's a Scala syntax.

[kgpai]

cc: @pedroerp @mbasmanova

[pedroerp]

We had discussions in the past about creating Java/JNI bindings, but I don't think anyone has actually done the work. Agreed it would be useful as more Java engines try to leverage Velox.

The one complication is that Velox doesn't really have a single public API (rather, there are multiple APIs working in different layers), so I suppose the first discussion would be to agree on the exact APIs we would like to expose. It would also be nice if we made them (at least mostly) compatible across C++ and pyVelox also.

[aditi-pandit]

+1 For the Velox API discussion. At IBM we have lots of engineers writing functions and having a Java API would be super useful.

[assignUser]

I think the API topic (in general, not specifically for java) should be discussed in the next monthly? @pedroerp I currently don't see anything planned, when will the next one be?

[mbasmanova]

+1 For the Velox API discussion. At IBM we have lots of engineers writing functions and having a Java API would be super useful.

@aditi-pandit My understanding of the proposal is to to provide Java API for PlanNode.h, not for writing functions. @hn5092 Would you clarify?

[pedroerp]

I think the API topic (in general, not specifically for java) should be discussed in the next monthly? @pedroerp I currently don't see anything planned, when will the next one be?

True, would be a good topic to discuss on that forum. @sanumandla has the schedule; I think that event expired, but we should probably it bring to back to life.

[mbasmanova]

CC: @FelixYBW @rui-mo

[FelixYBW]

The most urgent java API usage from Gluten is the wrapper to access the RowVector. One major performance issue from Gluten today is the UDFs. Simple UDF can be easy to port with the help of LLM today. But as long as the UDF uses some java package like org.apache.http.client.utils.URIBuilder, it ends up to search or port the whole package.

If Velox has java API to access the rowvector data, we still need to port the row based UDF to columnar based but it's will be much more easier. Also we can improve performance by removing the R2C/C2R or similar logic.

[hn5092]

The most urgent java API usage from Gluten is the wrapper to access the RowVector. One major performance issue from Gluten today is the UDFs. Simple UDF can be easy to port with the help of LLM today. But as long as the UDF uses some java package like org.apache.http.client.utils.URIBuilder, it ends up to search or port the whole package.

If Velox has java API to access the rowvector data, we still need to port the row based UDF to columnar based but it's will be much more easier. Also we can improve performance by removing the R2C/C2R or similar logic.

+1 For the Velox API discussion. At IBM we have lots of engineers writing functions and having a Java API would be super useful.

@aditi-pandit My understanding of the proposal is to to provide Java API for PlanNode.h, not for writing functions. @hn5092 Would you clarify?

The Java API for plannode is certainly a very important and fundamental one, and it is also a set of APIs that we have implemented. However, in my imagination, Velox could do much more. For example, by passing a Java list to it, it could perform calculations directly, without the need for engines like Spark. It's like a high-performance collection framework. Or, similar to DuckDB, it could directly read Parquet files and then perform computations, because many Java programs actually need this functionality very much, and the current Java/Scala lambda expressions cannot provide such efficient operations.

[hn5092]

The most urgent java API usage from Gluten is the wrapper to access the RowVector. One major performance issue from Gluten today is the UDFs. Simple UDF can be easy to port with the help of LLM today. But as long as the UDF uses some java package like org.apache.http.client.utils.URIBuilder, it ends up to search or port the whole package.

If Velox has java API to access the rowvector data, we still need to port the row based UDF to columnar based but it's will be much more easier. Also we can improve performance by removing the R2C/C2R or similar logic.

Our design entails a mapping between the entire vector batch and the underlying C++ Velox vector, with direct data read and write operations via unsafe. This API suite currently supports all Spark types and has shown to be quite competitive in performance when tested against Spark's own vector benchmark. (From what I know, the performance of the dictionary type is not very good in vector batch implementations like Arrow in the open-source community.)

We have replaced Spark's ParquetFormat vector batch with our vector implementation to address the issue of Velox Parquet not supporting many types. This replacement only involved modifications to the batch code.

In terms of UDFs, we can develop a vector eval on the Spark side to carry out row-wise evaluation on the batch. Then, we can insert data into the Velox batch using our batch API.

[zhztheplayer]

@hn5092 Thank you for sharing. Are you working with Gluten or with your own solution?

We have replaced Spark's ParquetFormat vector batch with our vector implementation to address the issue of Velox Parquet not supporting many types.

This seems interesting. IIUC it's a pure Java implementation of ParquetFormat, and backed with Velox in-memory data format, is that correct? Did you try comparing this ParquetFormat with Velox's native Parquet reader / writer in regard of performance?

Furthermore, I think we are basically discussing on both Velox query plan and Velox data format in the topic. Perhaps the two can be split. My personal feeling is implementing a plan API could anyway help on Java + Velox development. For data format, perhaps we can figure out the relationship between the new Java Velox format and existing Velox + Arrow solution at the beginning. Sharing data between Java and C++ may require for a well designed Velox C ABI, which could functionally overlap with Velox Arrow bridge a little bit. Do you have some thoughts on this part?

[FelixYBW]

In terms of UDFs, we can develop a vector eval on the Spark side to carry out row-wise evaluation on the batch. Then, we can insert data into the Velox batch using our batch API.

We already had the vector based UDFs in Gluten. current solution is to convert to Arrow format, then use Arrow's java API to access the data. This part does can be upstream to Spark but not much useful because Spark only has parquet scan output the columnar format.

The initial plan in Velox is that we can should have data level consistency so we can get one data format, two wrappers (Arrow's record batch and Velox's rowvector). But it doesn't work well yet.

@pedroerp should we add velox dataformat's java wrapper or should we leverage Arrow's java wrapper?

[FelixYBW]

Furthermore, I think we are basically discussing on both Velox query plan and Velox data format in the topic.

Data format one is more important to Gluten. Gluten needn't the Velox query plan's java wrapper.

[FelixYBW]

@hn5092 you may move the topic to discussion

[hn5092]

@hn5092 Thank you for sharing. Are you working with Gluten or with your own solution?

We have replaced Spark's ParquetFormat vector batch with our vector implementation to address the issue of Velox Parquet not supporting many types.

This seems interesting. IIUC it's a pure Java implementation of ParquetFormat, and backed with Velox in-memory data format, is that correct? Did you try comparing this ParquetFormat with Velox's native Parquet reader / writer in regard of performance?

Furthermore, I think we are basically discussing on both Velox query plan and Velox data format in the topic. Perhaps the two can be split. My personal feeling is implementing a plan API could anyway help on Java + Velox development. For data format, perhaps we can figure out the relationship between the new Java Velox format and existing Velox + Arrow solution at the beginning. Sharing data between Java and C++ may require for a well designed Velox C ABI, which could functionally overlap with Velox Arrow bridge a little bit. Do you have some thoughts on this part?

1. Performance-wise
   Velox's Parquet performance is definitely better than Spark's, but in terms of compatibility, data types, and filesystems (such as HDFS), Spark's native Parquet is somewhat superior. I was compelled to choose this solution initially (because Velox does not support timestamp types). Our current approach can also directly support other types, such as ORC, etc. At the same time, we also have a columnar distributed in-memory cache (cache Velox batch), so we haven't considered Velox's Parquet for the time being. If possible, I might port over Impala's Parquet later on, which is the world's fastest and has the best compatibility.

2. We gave up on Arrow
   We previously tried using Gluten and also attempted to implement our API on Arrow, but the performance was not satisfactory. Based on our past experience using Impala as the backend (we had also written a set of Java APIs for Impala rowbatch), we decided to write our own Velox batch Java API. Of course, our implementation is still a bit away from being contributable; constant dictionary types are not yet supported (currently, when we encounter this type, we just flatten it).

Additionally, I haven't yet seen any necessary benefits that Arrow provides for Java APIs. If there are advantages that I am not aware of, you can inform me.

[aditi-pandit]

@mbasmanova : From the IBM use-cases that I'm familiar with are close to what BinWei described above. The existing functions (scalars or aggregates) are written in Java and we want to wrap the row by row logic (or some functions could be columnar as well) within a Velox operator that can be invoked through SQL.

Remote UDF can be used for this also, though the Java API would be more performant.

This issue has been revised and rewritten few times, but when I commented that was my line of thought.

Velox PlanNode API came up in conversation few times when we discussed PlanValidation of native plans at the co-ordinator side.

[hn5092]

I hope this matter won't be put on hold. In the next few weeks, I'll take the time to submit a draft of the Velox Vector Java API for everyone to review. Please take a look and provide your own feedback. Upon completion, we will proceed with Phase 2, which involves the development of the Velox Node Plan Java API.

[hn5092]

Sharing the API will help plan the MVP and the roadmap.

I think we can first define the C++ API, and then have the implementations in other languages all uniformly interface with this layer of API. The current planbuilder is a very good interface. It can already directly implement the construction of plans, but it lacks an execution API. During execution, different scans need different parameters (such as for parquet/arrow stream), and this interface is quite challenging to define. I'm not sure if you have any better ideas.

[mbasmanova]

The current planbuilder is a very good interface ... but it lacks an execution API

@hn5092 The execution API would be something like AssertQueryBuilder (velox/exec/tests/utils/AssertQueryBuilder.h) which allows to provide splits (scan parameters).

Both PlanBuilder and AssertQueryBuilder are test-only utilities. They are super convenient to use and allow easily create a plan and run it. PlanNoteStats.h helpers then allow to get EXPLAIN ANALYZE-like output, which is great for debugging: https://facebookincubator.github.io/velox/develop/debugging/print-plan-with-stats.html

The killer feature of the PlanBuilder is ability to specify expressions using SQL. Under the covers, PlanBuilder uses DuckDB SQL parser (based on Postgres). Velox proper doesn't provide SQL API and doesn't include SQL parser.

Any user-oriented dataframe API must support SQL for expressions, otherwise, it is unusable.

Hence, the question is can we build a frontend to Velox that supports dataframes and SQL for expressions?

[hn5092]

thx @mbasmanova , i will see it

[hn5092]

@majetideepak

This is the JAVA API I've compiled based on our usage here, which includes logic for building execution plans, converting expressions, and execution. We are currently using JSON to pass all information for the sake of memory safety, though there might be better methods available in the future. Does anyone have any other thoughts or suggestions?

Java API Overview

1. Base Components

• Data types similar to Spark's DataType
• Results are handled using VeloxBatch
• API functions resemble Spark's columnarBatch

2. Expression Support

• Accepts string expressions like "a = 1" or `spark/flink Expression
• Outputs are currently in JSON format

public class VeloxExpression {
    public String parse(String expressionString, DataType dataType);

    // May be unnecessary 
    public String convert(Expression sparkOrFlinkExpression);  
}

3. Plan Construction

To build an execution plan, use the VeloxPlanBuilder class:

public class VeloxPlanBuilder {

    public void parquet(DataType dataType);

    public void iterator(DataType dataType);

    public void project(String[] projections);

    // ... additional methods similar to C++ VeloxPlanBuilder

    public String[] scanNodes();

    public String currentNodeId();

    public String planJson();
}

4. Plan Execution

Execute a plan using the VeloxExecution class:

public class VeloxExecution {

    public void initialize(String planJson);

    public void addIteratorSplit(String[] scanPlanIds, Iterator[] iterators);

    public void addParquetSplit(String[] scanPlanIds, Split[] splits);

    public void noMoreSplits();

    public boolean hasNext();

    public VeloxColumnarBatch next();

    public void close();
}

Usage Example

Here's how you can set up and use the Velox plan builder and execution classes:

VeloxPlanBuilder planBuilder = new VeloxPlanBuilder()
    .iterator(schema)
    .project(new String[] {"a", "b"});

String planJSON = planBuilder.planJson();

try(VeloxExecution execution = new VeloxExecution(planJSON)) {
execution.addIteratorSplit(planBuilder.scanNodes(), splits)
execution.noMoreSplits();
while(execution.hasNext()){
  //process batch
execution.next();
}
}

[pedroerp]

@hn5092 thanks for taking a stab at it. I sent you an invite to our Slack, it might be easier to discuss these things there first. We have also been thinking about the APIs we should provide in both C++ and Python.

[pedroerp]

The Java API for plannode is certainly a very important and fundamental one, and it is also a set of APIs that we have implemented. However, in my imagination, Velox could do much more. For example, by passing a Java list to it, it could perform calculations directly, without the need for engines like Spark. It's like a high-performance collection framework. Or, similar to DuckDB, it could directly read Parquet files and then perform computations, because many Java programs actually need this functionality very much, and the current Java/Scala lambda expressions cannot provide such efficient operations.

I'm +1 for adding bindings to the main APIs we need, but we need to be careful about the goals of the project. Velox is not meant to be a user facing data processing library by itself; it is not meant to be an alternative for things like Pandas, numpy, DuckDB and similar. The idea is that such projects could be built by leveraging Velox.

So, in my opinion, if the goal is to provide a very thin wrapper to expose Velox's main API in Java, this would be part of Velox (like PyVelox, although that is still a prototype). If the plan is to provide an API for end users (data scientists and such) so that they could use Velox's power to build data pipelines (read parquet files, aggregate data, similar to Pandas), I still think it's a very useful and compelling thing to be built, but it sounds like a different project whatsoever.

[pedroerp]

Now, regarding the actual APIs to be exposed. In my view, the main value Velox provides is a way to very efficiently execute query plans. In this sense, what an external API should comprise, in this order:

1. The query plan representation itself (plan nodes).
2. A way to execute it (tasks).

It happens that many plan nodes take expressions, so we will also need:

3. A way to represent expressions (expression nodes). One could argue that this is already part of 1.

Optionally, we could expose through the API a way to evaluate expressions as a standalone API independent of query plans:

4. Expression evaluation (exprSet).

Similarly, some of this API will rely on the actual data representation, so we will probably need to expose an API for Vectors (similar to Arrow).

5. Data representation API (vectors, types)

I think it even make sense to tackle things in this order (although 1. is arguably the hardest to expose), since Velox's main value is to execute plans, not to represent data (the latter could also be done using something like Arrow).

@mbasmanova @xiaoxmeng @oerling @Yuhta @majetideepak , thoughts?

[majetideepak]

I feel the Data representation API is independent. The query plan/functions could be hardcoded or auto generated for example.

[pedroerp]

Query plans and functions are different. Query plans are the main API in Velox (the whole purpose of the project is to execute query plans), while functions (scalar, aggregates and window), types, operators, etc are ways to extend the library behavior. One can think of the data representation as an implementation detail, although it does leak through the API.

[FelixYBW]

@pedroerp We are using Arrow's Java API to access the data. There are two issues observed:

1. Arrow format is still different from Velox's. The overhead of Arrow2Velox and Velox2Arrow is still unignorable
2. Arrow's Java binding overhead is also unignorable which needs some optimization.

@jinchengchenghh

[jinchengchenghh]

For Velox to Arrow, there is two steps:

1. export Vector to C++ struct ArrowArray, this is less than half time vs row to column for time int.
2. ArrowArray export to java Spark ColumnarBatch, there is FieldNode conversion and ValueVector generator, it is high.
   The total cost is nearly to optimized row to column. https://github.com/facebookincubator/velox/pull/10797/files
   I test it long time ago, so the number may be not very precise but Arrow to Velox is really an expensive operation.

@hn5092 also mentioned it is expensive.

We previously tried using Gluten and also attempted to implement our API on Arrow, but the performance was not satisfactory.

[FelixYBW]

what's the efficient way to implement java wrapper for StringView?

[hn5092]

If we use arrow to store parquet data on the java side, the performance bottleneck is dictionary reuse

[pedroerp]

What @majetideepak said. I don't remember seeing StringView support in Arrow java yet, so this may be part of the issue.

From @jinchengchenghh 's message, it seems like the cost is not necessarily from Velox to Arrow, but what you do later with this Arrow payload.

[ViggoC]

StringView is alrealy implemented in Java version. See ViewVarCharVector.
And RunEndVector is on the way. apache/arrow#43888

[hn5092]

I apologize for the inconvenience, as I have been busy organizing code and dealing with other matters. Today, I open-sourced one of our implementations, which, unfortunately, is based on Spark. Everyone is welcome to use it as a reference. This implementation may not be ideal; it is merely meant to offer a concept. My ultimate goal is for us all to adopt a unified and high-performance API, which would also help to reduce the cost of rebasing each time.

The entire implementation is fundamentally designed around the Velox API.

1. Zero-copy vectors:

   • Velox: https://github.com/proxverse/velox/tree/starry/velox/sdk/velox/vector
   • Java: https://github.com/proxverse/StarryDB/tree/main/starry-core/src/main/scala/org/apache/spark/sql/execution/columnar

2. SDK for plan building:

   • Velox: https://github.com/proxverse/velox/tree/starry/velox/sdk/velox/plan
   • Java: https://github.com/proxverse/StarryDB/blob/main/starry-core/src/main/scala/org/apache/spark/sql/execution/columnar/jni/NativePlanBuilder.java

3. The conversion of expressions:

   • Velox: https://github.com/proxverse/velox/tree/starry/velox/sdk/velox/expression
   • Java: https://github.com/proxverse/StarryDB/tree/main/starry-core/src/main/scala/org/apache/spark/sql/execution/columnar/expressions

4. Automatic exposure of Velox functions. Moving forward, we plan replace 3 to this path: https://github.com/proxverse/StarryDB/tree/main/starry-core/src/main/scala/org/apache/spark/sql/catalog.

[hn5092]

json is used to make memory safer

[zhztheplayer]

Thanks @hn5092.

And am I getting something wrong that, the JNI bridge contains one method for the creation of each plan node type? Any consideration not using a unified serde-based API?

[hn5092]

Thanks @hn5092.

And am I getting something wrong that, the JNI bridge contains one method for the creation of each plan node type? Any consideration not using a unified serde-based API?

We did not use JSON at the plan node level. The JSON for our entire plan is generated at the end. During the assembly of the plan, we maintained a pointer to the plan builder. JSON is mainly used for the expression and field sections, as they are more complex. Initially, our design for expressions used pointers, but we later found that it could easily lead to a core dump if not careful. For safety reasons, we ultimately decided to represent expressions in JSON.

[zhztheplayer]

We did not use JSON at the plan node level. The JSON for our entire plan is generated at the end. During the assembly of the plan, we maintained a pointer to the plan builder.

I see. But what's the blocker so far against using JSON at plan level? Say ideally we can have a Java-side plan builder to build the plan, then pass the whole serialized JSON though JNI.

[hn5092]

我们没有在计划节点级别使用 JSON。整个计划的 JSON 是在最后生成的。在计划的组装过程中，我们维护了一个指向计划构建器的指针。

我明白了。但目前在计划级别使用 JSON 的障碍是什么？理想情况下，我们可以有一个 Java 端计划构建器来构建计划，然后通过 JNI 传递整个序列化的 JSON。

Certainly, the best approach would be to generate JSON entirely on the Java side, which is one of the methods I would prefer. However, I have encountered some issues in the past:

1. The amount of work required for a comprehensive implementation: If we need to implement a full Java-based solution for producing JSON, it's not just about considering the plan nodes; we also need to take expressions into account. Otherwise, would we still rely on JNI for passing expressions? This leads to a lack of overall uniformity. Moreover, it's worth noting that the workload for serializing expressions is tremendous, comparable to writing a new version of Substrait.

2. The stability of the velox C++ API: In the year and more that I've been using velox, there have been changes to the API. If we were to implement a complete set on the Java side, we might have to maintain it concurrently, which would put certain pressure on maintenance.

The issues mentioned above are the reasons why I chose to use JNI to call and ultimately produce this JSON.