[VL] The POC of supporting Flink in Gluten #8839

shuai-xu · 2025-02-27T02:05:41Z

What changes were proposed in this pull request?

This pr is the Java side of POC to support Flink. It generates a GlutenCalOperator to run filter using native.
The draft design is here.

github-actions · 2025-02-27T02:05:59Z

Thanks for opening a pull request!

Could you open an issue for this pull request on Github Issues?

https://github.com/apache/incubator-gluten/issues

Then could you also rename commit message and pull request title in the following format?

[GLUTEN-${ISSUES_ID}][COMPONENT]feat/fix: ${detailed message}

See also:

Other pull requests

PHILO-HE · 2025-02-27T10:09:03Z

@shuai-xu, thanks for your great work! Could you draft a design doc? Google doc is preferred.

zhztheplayer · 2025-02-27T10:21:22Z

gluten-flink/core/src/main/java/org/apache/gluten/backendsapi/FlinkBackend.java

+ */
+package org.apache.gluten.backendsapi;
+
+public class FlinkBackend {


Worth emphasizing that Flink will not be a backend in Gluten. It's more considered a frontend or a framework or so.

OK, We can discuss it.

jinchengchenghh · 2025-02-27T11:20:07Z

Can we use gluten-substrait module? Looks like there exists too much duplicated code.

FelixYBW · 2025-02-27T19:46:27Z

Thank you for the PR! It eventually starts

majetideepak · 2025-02-27T20:44:31Z

Is there a design for this support for people who are not very familiar with Flink?

weiting-chen · 2025-02-28T02:25:49Z

Is there a design for this support for people who are not very familiar with Flink?

This is an initial PR and require more people to join with us for reviewing the design.
For the background and current potential issues, please check the link: https://docs.google.com/document/d/1VNMs1oR0c02kuQLFLQXZeTyk3CoPjrtAfiKGoLGxGzE/edit?usp=sharing

shuai-xu · 2025-02-28T03:22:01Z

Can we use gluten-substrait module? Looks like there exists too much duplicated code.

Yes, we need to discuss whether it need to translate to substrait plan or just call velox jni interface, if do need to translate to substrait plan, It need to reconstruct the gluten-substraint module a little to share it between gluten spark and flink.

shuai-xu · 2025-02-28T03:25:57Z

@PHILO-HE @majetideepak OK, I will write a design soon.

zhztheplayer · 2025-02-28T03:32:55Z

@PHILO-HE @majetideepak OK, I will write a design soon.

Hi @weiting-chen, would you share who is the author of https://docs.google.com/document/d/1VNMs1oR0c02kuQLFLQXZeTyk3CoPjrtAfiKGoLGxGzE/edit?usp=sharing?

I am opening a GitHub side discussion here: #8849

weiting-chen · 2025-02-28T03:38:22Z

@PHILO-HE @majetideepak OK, I will write a design soon.

Hi @weiting-chen, would you share who is the author of https://docs.google.com/document/d/1VNMs1oR0c02kuQLFLQXZeTyk3CoPjrtAfiKGoLGxGzE/edit?usp=sharing?

I am opening a GitHub side discussion here: #8849

Yes, done to add the author.

...flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenCalOperator.java

+        }
+    }
+
+    private native long nativeProcessElement(int executor, long data);


zjuwangg · 2025-03-03T08:55:55Z

Very glad to see such exciting progress on Flink support.
Looking forward on how to run the poc code in test...

ParyshevSergey · 2025-03-04T04:09:38Z

System.load("/home/xushuai/gluten/cpp-flink/build/libgflink.so");
How to get this library? Or which steps need to pass for compile it?

zhztheplayer · 2025-03-04T04:31:38Z

...flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenCalOperator.java

+                new BoundSplit(
+                        "5",
+                        -1,
+                        new ExternalStreamConnectorSplit("escs1", es.id())));


Would use connector-external-stream as connector ID.

It is currently a fixed value in Velox4j; https://github.com/velox4j/velox4j/blob/434ae37dfc3d5fb79788fe5bce41e41dd17901b5/src/main/cpp/main/velox4j/init/Init.cc#L94-L97

zhztheplayer · 2025-03-04T04:34:13Z

...flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenCalOperator.java

+        ExternalStream es = session.externalStreamOps().bind(new DownIterator(inputIterator));
+        List<BoundSplit> splits = List.of(
+                new BoundSplit(
+                        "5",


The planNodeId is the same with leaf scan node's node ID so Velox knows we bind this split to the scan. Perhaps we can pass the scan node ID into GlutenCalOperator somehow?

zhztheplayer · 2025-03-04T04:41:25Z

gluten-flink/planner/src/main/java/org/apache/gluten/rexnode/RexNodeConverter.java

+            return new ConstantTypedExpr(
+                    toType(literal.getType()),
+                    toVariant(literal),
+                    null);


Could use ConstantTypedExpr.create(toVariant(literal))

The create will call native method, I think we'd better not call native in client side, so we need not load native libraries in Flink client side.

zhztheplayer · 2025-03-04T04:44:06Z

...nner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecCalc.java

+        // add a mock input as velox not allow the source is empty.
+        PlanNode mockInput = new ValuesNode(
+                String.valueOf(ExecNodeContext.newNodeId()),
+                "",
+                false,
+                1);


Adding a scan node so we can bind the split to it. E.g.,

final TableScanNode scanNode = new TableScanNode( "id-1", ...(type), new ExternalStreamTableHandle("connector-external-stream"), List.of() );

zhztheplayer · 2025-03-04T04:45:36Z

gluten-flink/planner/src/main/java/org/apache/gluten/rexnode/RexNodeConverter.java

+            case VARCHAR:
+                return new VarCharValue(literal.getValue().toString());
+            case BINARY:
+                return new VarBinaryValue(literal.getValue().toString());


Should we use VarBinaryValue.create() to pass a byte array in?

ParyshevSergey · 2025-03-05T10:04:55Z

After try to run this POC, got error
Caused by: io.github.zhztheplayer.velox4j.exception.VeloxException: Exception: VeloxUserError Error Source: USER Error Code: INVALID_ARGUMENT Reason: Splits can be associated only with leaf plan nodes which require splits. Plan node ID 5 doesn't refer to such plan node.

shuai-xu · 2025-03-06T03:16:01Z

After try to run this POC, got error Caused by: io.github.zhztheplayer.velox4j.exception.VeloxException: Exception: VeloxUserError Error Source: USER Error Code: INVALID_ARGUMENT Reason: Splits can be associated only with leaf plan nodes which require splits. Plan node ID 5 doesn't refer to such plan node.

You can try the latest code.

ParyshevSergey · 2025-03-06T07:36:54Z

After try to run this POC, got error Caused by: io.github.zhztheplayer.velox4j.exception.VeloxException: Exception: VeloxUserError Error Source: USER Error Code: INVALID_ARGUMENT Reason: Splits can be associated only with leaf plan nodes which require splits. Plan node ID 5 doesn't refer to such plan node.

You can try the latest code.

Thanks for reply. I rerun with last commit and gotcha

Caused by: io.github.zhztheplayer.velox4j.exception.VeloxException: com.fasterxml.jackson.databind.exc.UnrecognizedPropertyException: Unrecognized field "sources" (class io.github.zhztheplayer.velox4j.plan.TableScanNode), not marked as ignorable (4 known properties: "id", "tableHandle", "outputType", "assignments"])
 at [Source: UNKNOWN; byte offset: #UNKNOWN] (through reference chain: io.github.zhztheplayer.velox4j.plan.TableScanNode["sources"])

I looked at TableScanNode and notice that json getter exist on "sources" but this field is absence. I try rebuild velox4j with Serde change (add disable for jackson DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES) and got

Caused by: io.github.zhztheplayer.velox4j.exception.VeloxException: couldn't find key name in dynamic object
	at io.github.zhztheplayer.velox4j.jni.JniWrapper.executeQuery(Native Method)
	at io.github.zhztheplayer.velox4j.jni.JniApi.executeQuery(JniApi.java:52)
	at io.github.zhztheplayer.velox4j.query.Queries.execute(Queries.java:14)
	at org.apache.gluten.table.runtime.operators.GlutenCalOperator.processElement(GlutenCalOperator.java:97)

SQL query

CREATE TABLE srcTbl (id INT, price INT, name STRING) WITH ('connector'='datagen');
CREATE TABLE snkTbl (id INT, price INT) WITH ('connector'='blackhole');
INSERT INTO snkTbl SELECT id, price FROM srcTbl WHERE price > 10;

shuai-xu · 2025-03-07T02:07:18Z

@ParyshevSergey , your query runs well in my side, could you give more logs?

zhztheplayer · 2025-03-07T06:03:02Z

After try to run this POC, got error Caused by: io.github.zhztheplayer.velox4j.exception.VeloxException: Exception: VeloxUserError Error Source: USER Error Code: INVALID_ARGUMENT Reason: Splits can be associated only with leaf plan nodes which require splits. Plan node ID 5 doesn't refer to such plan node.

You can try the latest code.

Thanks for reply. I rerun with last commit and gotcha
Caused by: io.github.zhztheplayer.velox4j.exception.VeloxException: com.fasterxml.jackson.databind.exc.UnrecognizedPropertyException: Unrecognized field "sources" (class io.github.zhztheplayer.velox4j.plan.TableScanNode), not marked as ignorable (4 known properties: "id", "tableHandle", "outputType", "assignments"])
 at [Source: UNKNOWN; byte offset: #UNKNOWN] (through reference chain: io.github.zhztheplayer.velox4j.plan.TableScanNode["sources"])
I looked at TableScanNode and notice that json getter exist on "sources" but this field is absence. I try rebuild velox4j with Serde change (add disable for jackson DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES) and got
Caused by: io.github.zhztheplayer.velox4j.exception.VeloxException: couldn't find key name in dynamic object
	at io.github.zhztheplayer.velox4j.jni.JniWrapper.executeQuery(Native Method)
	at io.github.zhztheplayer.velox4j.jni.JniApi.executeQuery(JniApi.java:52)
	at io.github.zhztheplayer.velox4j.query.Queries.execute(Queries.java:14)
	at org.apache.gluten.table.runtime.operators.GlutenCalOperator.processElement(GlutenCalOperator.java:97)
SQL query
CREATE TABLE srcTbl (id INT, price INT, name STRING) WITH ('connector'='datagen');
CREATE TABLE snkTbl (id INT, price INT) WITH ('connector'='blackhole');
INSERT INTO snkTbl SELECT id, price FROM srcTbl WHERE price > 10;

The error is likely because this velox4j change velox4j/velox4j@b1e66a2 is not included in the release yet. I will release a newer version of velox4j then let you know.

zhztheplayer · 2025-03-07T06:04:11Z

Or you can try build velox4j by your self:

https://github.com/velox4j/velox4j?tab=readme-ov-file#prerequisites
https://github.com/velox4j/velox4j?tab=readme-ov-file#build-from-source

ParyshevSergey · 2025-03-07T06:13:08Z

@ParyshevSergey , your query runs well in my side, could you give more logs?

log.txt

I attache the log. Also add some information, in OpenSource Flink 1.20 i create module flink-examples-gluten with all needed dependencies, in this module i run the job.

package org.apache.flink.examples.gluten;

import io.github.zhztheplayer.velox4j.Velox4j;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.util.concurrent.ExecutionException;

public class TestJob {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Velox4j.initialize();
        Configuration conf = new Configuration();
        StreamExecutionEnvironment see =
                StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);
        see.setParallelism(1);
        StreamTableEnvironment ste = StreamTableEnvironment.create(see);

        ste.executeSql(
                "create table srcTbl (id int, price int, name string) WITH ('connector'='datagen')");
        ste.executeSql("create table snkTbl (id int, price int) WITH ('connector'='blackhole')");

        ste.executeSql("insert into snkTbl select id, price from srcTbl where price > 10").await();
    }
}

ParyshevSergey · 2025-03-07T06:58:51Z

Or you can try build velox4j by your self:

https://github.com/velox4j/velox4j?tab=readme-ov-file#prerequisites https://github.com/velox4j/velox4j?tab=readme-ov-file#build-from-source

Thanks for reply, I build locally velox4j and rerun test

log.txt

PHILO-HE

Some superficial comments. Thank you!

PHILO-HE · 2025-03-10T17:08:55Z

docs/get-started/Flink.md

I prefer firstly putting this doc under gluten-flink/docs folder as it is an experimental and separate feature.

PHILO-HE · 2025-03-11T08:35:05Z

docs/get-started/Flink.md

+**For x86_64**
+
+```bash
+## make sure jdk8 is used


The comment is not aligned with the following.

PHILO-HE · 2025-03-11T08:35:18Z

docs/get-started/Flink.md

+**For aarch64**
+
+```bash
+## make sure jdk8 is used


PHILO-HE · 2025-03-11T09:45:53Z

gluten-flink/runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java

+    }
+
+    private void startTaskManagerRunnerServices() throws Exception {
+        synchronized (lock) {


Redundant synchronized, since it has been added on calling side. Assume it is only called there.

This is flink code, I just copy it and add some code to load libvelox.so. I think we can ignore it first.

PHILO-HE · 2025-03-11T09:48:13Z

gluten-flink/runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java

+    }
+
+    private CompletableFuture<Void> shutDownServices() {
+        synchronized (lock) {


PHILO-HE · 2025-03-11T10:52:42Z

gluten-flink/runtime/src/main/java/org/apache/gluten/vectorized/FlinkRowToVLRowConvertor.java

+import java.util.List;
+
+/** Converter between velox RowVector and Flink RowData. */
+public class FlinkRowToVLRowConvertor {


->FlinkRowToVeloxColumnarConvertor

VLRow is misleading.

PHILO-HE · 2025-03-11T10:56:35Z

gluten-flink/pom.xml

+  <modules>
+    <module>planner</module>
+    <module>loader</module>
+    <module>runtime</module>


Seems loader is just for loading classes? Could you explain why we need a dedicated module for it?

As I say in the design doc. Flink use a loader to load the modules only used in client side. This is to avoid class confilicting when submit multi jobs by one client. We just follow flink in gluten.

yangxiao0320 · 2025-03-12T08:53:36Z

When will stateful operators be supported?

zhztheplayer · 2025-03-13T16:21:33Z

@ParyshevSergey

I've published a new SNAPSHOT jar that may relate to this. Would you like to try again (with mvn -U)? Thanks.

See the published SNAPSHOT jar list:

https://central.sonatype.com/service/rest/repository/browse/maven-snapshots/io/github/zhztheplayer/velox4j/0.1.0-SNAPSHOT/

zhztheplayer · 2025-03-13T16:26:50Z

@shuai-xu

Non-blocking style execution is now working since the PR was finished. With the feature, Velox will not stop processing data until we explicitly notified the task. This could remove a major blocker against implementing filters with Velox.

Would you also help verify it from this PR's end? Thank you.

ParyshevSergey · 2025-03-14T10:28:00Z

@ParyshevSergey

I've published a new SNAPSHOT jar that may relate to this. Would you like to try again (with mvn -U)? Thanks.

See the published SNAPSHOT jar list:

https://central.sonatype.com/service/rest/repository/browse/maven-snapshots/io/github/zhztheplayer/velox4j/0.1.0-SNAPSHOT/

I build velox4j and remove line with drivers check in QueryExecutor.cc(drivers_empty() check) and then POC started up normally.

zhztheplayer · 2025-03-14T17:18:23Z

I build velox4j and remove line with drivers check in QueryExecutor.cc(drivers_empty() check) and then POC started up normally.

Glad to hear that. Does the filter-project routine work from your end?

ParyshevSergey · 2025-03-17T03:12:00Z

I build velox4j and remove line with drivers check in QueryExecutor.cc(drivers_empty() check) and then POC started up normally.

Glad to hear that. Does the filter-project routine work from your end?

Filter and projection working well. How important check of "drivers_empty()" in QueryExecutor.cc ?

shuai-xu · 2025-03-18T03:53:44Z

When will stateful operators be supported?

This is a long term work, may be in latter half of this year.

zhztheplayer · 2025-03-18T09:28:05Z

I build velox4j and remove line with drivers check in QueryExecutor.cc(drivers_empty() check) and then POC started up normally.

Glad to hear that. Does the filter-project routine work from your end?

Filter and projection working well. How important check of "drivers_empty()" in QueryExecutor.cc ?

I think it's not really important and we can remove it anyway.

oxwangfeng125 · 2025-03-19T06:57:57Z

@ParyshevSergey

I've published a new SNAPSHOT jar that may relate to this. Would you like to try again (with mvn -U)? Thanks.

See the published SNAPSHOT jar list:

https://central.sonatype.com/service/rest/repository/browse/maven-snapshots/io/github/zhztheplayer/velox4j/0.1.0-SNAPSHOT/

I find some class is not in that jar，please update the new jar ，Thanks

zhztheplayer · 2025-03-19T14:27:04Z

@ParyshevSergey
I've published a new SNAPSHOT jar that may relate to this. Would you like to try again (with mvn -U)? Thanks.
See the published SNAPSHOT jar list:
https://central.sonatype.com/service/rest/repository/browse/maven-snapshots/io/github/zhztheplayer/velox4j/0.1.0-SNAPSHOT/

I find some class is not in that jar，please update the new jar ，Thanks

The latest snapshot https://central.sonatype.com/service/rest/repository/browse/maven-snapshots/io/github/zhztheplayer/velox4j/0.1.0-SNAPSHOT/0.1.0-20250316.095536-16/ should already contain the newest code.

Thought this PR might still rely on old versions of Velox4J so need to be updated. Could you share the error message you saw?

lsyldliu · 2025-03-21T06:25:50Z

Very glad to see this great work has been pushing forward.

leonardBang · 2025-03-21T06:39:52Z

Cool, keep an eye on this PR

The POC of supporting Flink in Gluten

396abba

weiting-chen added the enhancement New feature or request label Feb 27, 2025

weiting-chen changed the title ~~The POC of supporting Flink in Gluten~~ [VL] The POC of supporting Flink in Gluten Feb 27, 2025

zhztheplayer reviewed Feb 27, 2025

View reviewed changes

refine the directories to follow flink stytle

e516d49

zjuwangg reviewed Mar 3, 2025

View reviewed changes

...flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenCalOperator.java Outdated

}

}

private native long nativeProcessElement(int executor, long data);

This comment was marked as resolved.

Sign in to view

FelixYBW mentioned this pull request Mar 4, 2025

Add initial flink support #8888

Open

shuai-xu added 2 commits March 4, 2025 12:23

use velox4j to execute

3afead5

add mising files

974b972

zhztheplayer reviewed Mar 4, 2025

View reviewed changes

zhztheplayer mentioned this pull request Mar 4, 2025

Make some essential constructors public velox4j/velox4j#80

Merged

Example run pass and add docs

4a15f3f

github-actions bot added the DOCS label Mar 6, 2025

PHILO-HE reviewed Mar 11, 2025

View reviewed changes

support multi operator chain in native

a6e9540

shuai-xu added 2 commits March 18, 2025 12:01

remove unnecessary files

f6721f1

fix header check

776f253

[VL] The POC of supporting Flink in Gluten #8839

Are you sure you want to change the base?

[VL] The POC of supporting Flink in Gluten #8839

Conversation

shuai-xu commented Feb 27, 2025 • edited Loading

What changes were proposed in this pull request?

github-actions bot commented Feb 27, 2025

PHILO-HE commented Feb 27, 2025

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jinchengchenghh commented Feb 27, 2025

FelixYBW commented Feb 27, 2025

majetideepak commented Feb 27, 2025 • edited Loading

weiting-chen commented Feb 28, 2025

shuai-xu commented Feb 28, 2025

shuai-xu commented Feb 28, 2025

zhztheplayer commented Feb 28, 2025

weiting-chen commented Feb 28, 2025

This comment was marked as resolved.

zjuwangg commented Mar 3, 2025

ParyshevSergey commented Mar 4, 2025

zhztheplayer Mar 4, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ParyshevSergey commented Mar 5, 2025

shuai-xu commented Mar 6, 2025

ParyshevSergey commented Mar 6, 2025 • edited Loading

shuai-xu commented Mar 7, 2025

zhztheplayer commented Mar 7, 2025

zhztheplayer commented Mar 7, 2025 • edited Loading

ParyshevSergey commented Mar 7, 2025

ParyshevSergey commented Mar 7, 2025

PHILO-HE left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

yangxiao0320 commented Mar 12, 2025

zhztheplayer commented Mar 13, 2025

zhztheplayer commented Mar 13, 2025

ParyshevSergey commented Mar 14, 2025

zhztheplayer commented Mar 14, 2025

ParyshevSergey commented Mar 17, 2025 • edited Loading

shuai-xu commented Mar 18, 2025

zhztheplayer commented Mar 18, 2025

oxwangfeng125 commented Mar 19, 2025

zhztheplayer commented Mar 19, 2025 • edited Loading

lsyldliu commented Mar 21, 2025

leonardBang commented Mar 21, 2025

shuai-xu commented Feb 27, 2025 •

edited

Loading

majetideepak commented Feb 27, 2025 •

edited

Loading

zhztheplayer Mar 4, 2025 •

edited

Loading

ParyshevSergey commented Mar 6, 2025 •

edited

Loading

zhztheplayer commented Mar 7, 2025 •

edited

Loading

ParyshevSergey commented Mar 17, 2025 •

edited

Loading

zhztheplayer commented Mar 19, 2025 •

edited

Loading