Flink + Bolt

[lgbo-ustc]

We have recently implemented a native execution engine for Flink based on Velox and have successfully run some Nexmark test cases. Currently, the system does not yet fully support stateful operators. Since progress on introducing new features in Velox has been relatively slow, we are considering whether Bolt could serve as an alternative implementation.

Flink execution follows a push model, while Velox execution is based on a pull model. During earlier development, we attempted to implement a push-like execution flow in Velox, but this would have required significant modifications to the framework. We are now exploring whether Flink’s computational characteristics can be met without changing the execution model of Velox/Bolt. After some reconsideration, we believe it is feasible. Below is the approach we have in mind, and we would like to hear feedback from the community.

We intend to treat Velox/Bolt as the internal computation implementation of a Flink operator. From this perspective, whether Velox/Bolt operates in pull or push mode is not fundamentally different—the two are equivalent. The internal execution flow within a Flink operator would be as follows:

public void processElement(StreamRecord<T> element) {
    /*
     * 1) Feed the input element into the source node of the Velox/Bolt execution plan
     * 2) Call getOutput to retrieve results from the Velox/Bolt execution plan
     * 3) Push the result to Flink’s output collector
     */
}

This implementation does not disrupt Flink’s push-based execution logic and preserves its backpressure control mechanism.

In Flink’s StreamGraphTranslator, we have modified the JobGraph. We still follow the principle that if two adjacent operators can both be offloaded to the native execution engine, they are merged into the same Velox/Bolt plan, forming a new operator. Currently, Velox/Bolt does not seem to support operators with multiple outputs, which may require an implementation similar to a local exchange queue. We have adopted a simpler approach: if an operator has multiple outputs, it will not be merged with its downstream operators, and communication between them will still occur through Flink channels.

The StreamGraphTranslator partitions an operator chain into different segments based on the following rules:

1. If two adjacent operators have different offloading capabilities to the native execution engine, they are placed in separate segments.

2. If an operator has multiple outputs, it is placed in a different segment from its downstream operators.

4. If an operator has multiple inputs, it is placed in a different segment from its upstream operators.

Changes to an operator chain before and after the modification are as follows.

We have also introduced a new serializer, RowVectorSerializer. When two adjacent Flink operators can both be offloaded, the channel between them is configured to use RowVectorSerializer, allowing direct RowVector data transfer and avoiding row-column conversion overhead. Moreover, within the same node, two operators will only pass a RowVector pointer without serialization or deserialization.

[zhanglistar]

@frankobe How do you think?

[frankobe]

It will be beneficial to join force on the Flink acceleration plan. RocksDB-based stateful operator is implemented internally. @zhanglistar @lgbo-ustc Feel free to comment in https://docs.google.com/document/d/1gNf-9VJEyMw1icEh3UnNsGwoGYuZu3qC3RIChWeS3F4/edit?tab=t.0#heading=h.34xj7176473v

@luozenglin and @yangzhg should we merge the #23 here?

[yangzhg]

should we merge the #23 here

OK. That should be a better way to cooperate with the community.

[liuyongvs]

hi @frankobe @lgbo-ustc what is the progress ?

[frankobe]

@liuyongvs We are rolling out Flink-on-Bolt (project: Blint) on stateless & stateful Flink SQL jobs internally. The current plan is to open source Blint as a subproject under Bolt on 26Q3 - 26Q4