Flink support

[zhztheplayer]

There is a PoC PR of Flink support recently submitted: #8839 (author: @shuai-xu)
The background doc: https://docs.google.com/document/d/1VNMs1oR0c02kuQLFLQXZeTyk3CoPjrtAfiKGoLGxGzE/edit?usp=sharing (author: @weiting-chen )
The design doc: WIP

There are no existing issues / discussions tracking on Flink support. So let's discuss here.

[zhztheplayer]

Thank everyone putting efforts on Flink support. Glad to see Flink experts willing to join.

For the implementation, my 2 cents so far is to extract a Java accelerator library out for our next framework support. I have been working on one https://github.com/velox4j/velox4j at which you may want to take a look. @shuai-xu

Substrait haven't been helping much on the extensibility of Gluten. We have to do a bunch of customizations on it that makes Gluten difficult to integrate with other Susbtrait consumer libraries. Moreover, Velox and CH both have incompatible usages on same Substrait features, for example the optimization / enhancement and so, which make it difficult to integrate a new framework with both VL / CH backends at the same time.

My feeling is a layered design will be the better arch for such accelerators. We build a Java library that tightly binds to the native libraries through JNI, then translate framework's query plan into library's query plan in Java in one shot. Things will be made much clearer and maintenance will be much simpler with such architecture.

[shuai-xu]

Velox4j is a good idea, it can simplify the usage of velox. I have some requirements about how to generate query plan and the data transfer between framework and velox. I will discuss with you offline.

[zhztheplayer]

@shuai-xu

In current version of Velox4j, a viable method to emulate trivial addInput / getOutput invocations is to build an ExternalStream with a custom iterator that is backed by a Queue<RowVector>. When you tend to call addInput, you just push a RowVector to the queue. Then you could call getOutput on the UpIterator to get next output RowVector from Velox.

This could be a simple solution at the moment and I will explore on better approaches soon.

[FelixYBW]

Substrait haven't been helping much on the extensibility of Gluten.

Gluten currently has the CK backend and Velox backend, they do share some logic when creating the substrait plan.

[FelixYBW]

How much code we can reuse from Gluten? if we can't reuse it may be a better idea to create a subproject. @weiting-chen

[zjuwangg]

Currently I think it is appropriate to keep it as a submodule. When it becomes more mature, we can discuss separating it into a subproject.
Since gluten is designed to support multiple computing frameworks, we can improve the design and make more abstractions to improve code reuse.

[zhanglistar]

Some Ideas:

1. Data Structure Adaptation: Convert RowData to Vector, supporting RowKind.

2. Operator Support: Extend Velox Operator to support state and rollback.

3. State Management: Integrate Flink's StateBackend into Velox.

4. Distributed Execution: Connect Flink's networking and scheduling logic to Velox.

5. Execution Flow: Replace Flink StreamTask with Velox Task.

6. State Engine

Velox does not have an internal state. We can introduce RocksDB as the state, which is not too difficult. However, this could become a potential performance bottleneck, so careful consideration is necessary. Since RocksDB involves disk operations, it is essential to orchestrate IO as asynchronous batch reads.

2. Stream retract Functionality

This is one of the fundamental differences between stream and batch processing. Streams are updated, while batches are not. For example, a table from MySQL naturally has updates and deletes. Flink introduces a stream retract mechanism to address this issue, which involves carrying metadata on the data to indicate whether it is an update, delete, or insert. Clearly, Velox does not have this capability.

Several modifications are needed:

1. Operators need to support rollback, such as window, stream join, stream aggregation, etc. This means that almost all operators will need to add this interface, which is a significant amount of work.
2. The data stream needs to include rollback indicators in the vector to indicate whether the current data is rollback data.

Once both the operators and the data stream are modified to support rollback, the overall execution flow should be fine.

3. Stateless Operators such as Map, Filter, FlatMap, KeyBy, Project/Calc, Union, Split/Select, and some Sources and Sinks do not depend on state and only transform the current input. They are suitable for simple data processing and can be reused directly. This is similar to the approach taken by Ant Group, which offloads expression calculations to Velox using JNI. However, due to potential row-column conversion and low coverage, there may be performance issues, and the overall speedup may not be significant.