You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@gluten.apache.org by ho...@apache.org on 2025/04/25 11:19:25 UTC
(incubator-gluten) branch main updated: [VL] The PoC of Flink support in Gluten (#8839)
This is an automated email from the ASF dual-hosted git repository.
hongze pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-gluten.git
The following commit(s) were added to refs/heads/main by this push:
new d4cd57eb3d [VL] The PoC of Flink support in Gluten (#8839)
d4cd57eb3d is described below
commit d4cd57eb3dc92b7c59608e99c77ab3908acd66da
Author: shuai.xu <xu...@bigo.sg>
AuthorDate: Fri Apr 25 19:19:21 2025 +0800
[VL] The PoC of Flink support in Gluten (#8839)
The POC to support Flink in Gluten. It generates a GlutenCalOperator to run filter in native. The draft design is [here](https://docs.google.com/document/d/1eC0VDTiqlrFJvkEt2sk__21gxaunCYF1RLdlMTaPOCk/edit?usp=sharing).
The PR is the initial part for adding an experimental Apache Flink accelerator backed by Velox to Gluten's code base, it's namely Gluten-Flink which will be considered part of the Apache Gluten's accelerator family.
In the PR, we are introducing a new top-level folder `gluten-flink` which accommodates all the the code Gluten-Flink will need. It's a individual Maven project than a sub-module of the existing Gluten-Spark work. This may be a temporary folder layout to be reorganized in the future.
This work may involve unexpected bugs and instabilities. We expect to deliver a stabler version with the subsequent PRs in the next few months .
---
gluten-flink/README.md | 7 +
gluten-flink/docs/Flink.md | 117 +++
gluten-flink/docs/data-generator.sql | 3 +
gluten-flink/loader/pom.xml | 82 ++
.../flink/table/planner/loader/PlannerModule.java | 200 +++++
gluten-flink/planner/pom.xml | 68 ++
.../plan/nodes/exec/stream/StreamExecCalc.java | 152 ++++
.../apache/gluten/rexnode/FunctionMappings.java | 44 ++
.../apache/gluten/rexnode/RexNodeConverter.java | 150 ++++
.../main/java/org/apache/gluten/rexnode/Utils.java | 49 ++
gluten-flink/pom.xml | 41 +
gluten-flink/runtime/pom.xml | 69 ++
.../apache/flink/client/StreamGraphTranslator.java | 168 +++++
.../runtime/taskexecutor/TaskManagerRunner.java | 823 +++++++++++++++++++++
.../LegacySourceTransformationTranslator.java | 127 ++++
.../translators/SinkTransformationTranslator.java | 556 ++++++++++++++
.../operators/GlutenOneInputOperatorFactory.java | 49 ++
.../streaming/api/operators/GlutenOperator.java | 28 +
.../api/operators/GlutenStreamSource.java | 55 ++
.../table/runtime/operators/GlutenCalOperator.java | 134 ++++
.../runtime/operators/GlutenSourceFunction.java | 99 +++
.../apache/gluten/util/LogicalTypeConverter.java | 62 ++
.../vectorized/FlinkRowToVLVectorConvertor.java | 105 +++
.../apache/gluten/vectorized/VLVectorIterator.java | 48 ++
24 files changed, 3236 insertions(+)
diff --git a/gluten-flink/README.md b/gluten-flink/README.md
new file mode 100644
index 0000000000..4c9a2ce023
--- /dev/null
+++ b/gluten-flink/README.md
@@ -0,0 +1,7 @@
+
+# Gluten Flink Project (Experimental)
+Gluten for Flink is an experimental feature, and it is under fast developing now,
+you can refer to [user guide](docs/Flink.md) for a quick usage.
+
+## Notes
+This first POC version may have memory leak, and leave tmp directories on the disk. We will fix it soon in the next version.
\ No newline at end of file
diff --git a/gluten-flink/docs/Flink.md b/gluten-flink/docs/Flink.md
new file mode 100644
index 0000000000..55ae5ea0ef
--- /dev/null
+++ b/gluten-flink/docs/Flink.md
@@ -0,0 +1,117 @@
+---
+layout: page
+title: Gluten For Flink with Velox Backend
+nav_order: 1
+---
+
+# Supported Version
+
+| Type | Version |
+|-------|------------------------------|
+| Flink | 1.19.2 |
+| OS | Ubuntu20.04/22.04, Centos7/8 |
+| jdk | openjdk11/jdk17 |
+| scala | 2.12 |
+
+# Prerequisite
+
+Currently, with static build Gluten+Flink+Velox backend supports all the Linux OSes, but is only tested on **Ubuntu20.04**. With dynamic build, Gluten+Velox backend support **Ubuntu20.04/Ubuntu22.04/Centos7/Centos8** and their variants.
+
+Currently, the officially supported Flink version is 1.19.2.
+
+We need to set up the `JAVA_HOME` env. Currently, Gluten supports **java 11** and **java 17**.
+
+**For x86_64**
+
+```bash
+## make sure jdk11 is used
+export JAVA_HOME=/usr/lib/jvm/java-11-openjdk-amd64
+export PATH=$JAVA_HOME/bin:$PATH
+```
+
+**For aarch64**
+
+```bash
+## make sure jdk11 is used
+export JAVA_HOME=/usr/lib/jvm/java-11-openjdk-arm64
+export PATH=$JAVA_HOME/bin:$PATH
+```
+
+**Get Velox4j**
+
+Gluten for Flink depends on [Velox4j](https://github.com/velox4j/velox4j) commit:7aaa6465cb2c56c8898b737bcc770bae35a94ee4 to call velox. This is an experimental feature.
+You need to get the Velox4j code, and compile it first.
+
+```bash
+## fetch velox4j code
+git clone https://github.com/velox4j/velox4j.git
+cd velox4j
+git reset --hard 7aaa6465cb2c56c8898b737bcc770bae35a94ee4
+mvn clean install
+```
+**Get gluten**
+
+```bash
+## config maven, like proxy in ~/.m2/settings.xml
+
+## fetch gluten code
+git clone https://github.com/apache/incubator-gluten.git
+```
+
+# Build Gluten Flink with Velox Backend
+
+```
+cd /path/to/gluten/gluten-flink
+mvn clean package
+```
+
+## Dependency library deployment
+
+You need to get the Velox4j packages and used them with gluten.
+Velox4j jar available now is velox4j-0.1.0-SNAPSHOT.jar.
+
+## Submit the Flink SQL job
+
+Submit test script from `flink run`. You can use the `StreamSQLExample` as an example.
+
+### Flink local cluster
+
+After deploying flink binaries, please add gluten-flink jar to flink library path,
+including gluten-flink-runtime-1.4.0.jar, gluten-flink-loader-1.4.0.jar and Velox4j jars above.
+And make them loaded before flink libraries.
+Then you can go to flink binary path and use the below scripts to
+submit the example job.
+
+```bash
+bin/start-cluster.sh
+bin/flink run -d -m 0.0.0.0:8080 \
+ -c org.apache.flink.table.examples.java.basics.StreamSQLExample \
+ lib/flink-examples-table_2.12-1.20.1.jar
+```
+
+Then you can get the result in `log/flink-*-taskexecutor-*.out`.
+And you can see an operator named `gluten-cal` from the web frontend of your flink job.
+
+#### All operators executed by native
+Another example supports all operators executed by native.
+You can use the data-generator.sql in dev directory.
+
+```bash
+bin/sql-client.sh -f data-generator.sql
+```
+
+### Flink Yarn per job mode
+
+TODO
+
+## Performance
+Using the data-generator example, it shows that for native execution, it can generate 10,0000
+records in about 60ms, while Flink generator 10,000 records in about 600ms. It runs 10 times faster.
+More perf cases to be added.
+
+## Notes:
+Now both Gluten for Flink and Velox4j have not a bundled jar including all jars depends on.
+So you may have to add these jars by yourself, which may including guava-33.4.0-jre.jar, jackson-core-2.18.0.jar,
+jackson-databind-2.18.0.jar, jackson-datatype-jdk8-2.18.0.jar, jackson-annotations-2.18.0.jar, arrow-memory-core-18.1.0.jar,
+arrow-memory-unsafe-18.1.0.jar, arrow-vector-18.1.0.jar, flatbuffers-java-24.3.25.jar, arrow-format-18.1.0.jar, arrow-c-data-18.1.0.jar.
+We will supply bundled jars soon.
diff --git a/gluten-flink/docs/data-generator.sql b/gluten-flink/docs/data-generator.sql
new file mode 100644
index 0000000000..9c520c4f2e
--- /dev/null
+++ b/gluten-flink/docs/data-generator.sql
@@ -0,0 +1,3 @@
+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;
diff --git a/gluten-flink/loader/pom.xml b/gluten-flink/loader/pom.xml
new file mode 100644
index 0000000000..bc32eedeac
--- /dev/null
+++ b/gluten-flink/loader/pom.xml
@@ -0,0 +1,82 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.gluten</groupId>
+ <artifactId>gluten-flink</artifactId>
+ <version>1.4.0-SNAPSHOT</version>
+ <relativePath>../pom.xml</relativePath>
+ </parent>
+
+ <artifactId>gluten-flink-loader</artifactId>
+ <name>Gluten Flink Loader</name>
+ <packaging>jar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-core</artifactId>
+ <version>${flink.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-planner_${scala.binary.version}</artifactId>
+ <version>${flink.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.gluten</groupId>
+ <artifactId>gluten-flink-planner</artifactId>
+ <version>${project.version}</version>
+ <scope>runtime</scope>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-dependency-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>copy-table-planner-jars</id>
+ <phase>prepare-package</phase>
+ <goals>
+ <goal>copy</goal>
+ </goals>
+ <configuration>
+ <artifactItems>
+ <artifactItem>
+ <groupId>org.apache.gluten</groupId>
+ <artifactId>gluten-flink-planner</artifactId>
+ <version>${project.version}</version>
+ <type>jar</type>
+ <overWrite>true</overWrite>
+ <destFileName>gluten-flink-planner.jar</destFileName>
+ </artifactItem>
+ </artifactItems>
+ <outputDirectory>${project.build.directory}/classes</outputDirectory>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+</project>
diff --git a/gluten-flink/loader/src/main/java/org/apache/flink/table/planner/loader/PlannerModule.java b/gluten-flink/loader/src/main/java/org/apache/flink/table/planner/loader/PlannerModule.java
new file mode 100644
index 0000000000..38f7b89c33
--- /dev/null
+++ b/gluten-flink/loader/src/main/java/org/apache/flink/table/planner/loader/PlannerModule.java
@@ -0,0 +1,200 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.loader;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.ConfigurationUtils;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.core.classloading.ComponentClassLoader;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.delegation.ExecutorFactory;
+import org.apache.flink.table.delegation.PlannerFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.util.FileUtils;
+import org.apache.flink.util.IOUtils;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.URL;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.UUID;
+import java.util.stream.Stream;
+
+/**
+ * This class will overwrite the PlannerModule in Flink to load gluten-flink-planner.jar.
+ * So that it will load the classes in gluten code first.
+ */
+class PlannerModule {
+
+ /**
+ * The name of the table planner dependency jar, bundled with flink-table-planner-loader module
+ * artifact.
+ */
+ static final String FLINK_TABLE_PLANNER_FAT_JAR = "flink-table-planner.jar";
+
+ private static final String HINT_USAGE =
+ "mvn clean package -pl flink-table/flink-table-planner,flink-table/flink-table-planner-loader -DskipTests";
+
+ private static final String[] OWNER_CLASSPATH =
+ Stream.concat(
+ Arrays.stream(CoreOptions.PARENT_FIRST_LOGGING_PATTERNS),
+ Stream.of(
+ // These packages are shipped either by
+ // flink-table-runtime or flink-dist itself
+ "org.codehaus.janino",
+ "org.codehaus.commons",
+ "org.apache.commons.lang3",
+ "org.apache.commons.math3",
+ // with hive dialect, hadoop jar should be in classpath,
+ // also, we should make it loaded by owner classloader,
+ // otherwise, it'll throw class not found exception
+ // when initialize HiveParser which requires hadoop
+ "org.apache.hadoop"))
+ .toArray(String[]::new);
+
+ private static final String[] COMPONENT_CLASSPATH = new String[] {
+ "org.apache.flink",
+ "org.apache.gluten",
+ "io.github",
+ "com.google"
+ };
+
+ private static final Map<String, String> KNOWN_MODULE_ASSOCIATIONS = new HashMap<>();
+
+ static {
+ KNOWN_MODULE_ASSOCIATIONS.put("org.apache.gluten.table.runtime", "gluten-flink-runtime");
+ KNOWN_MODULE_ASSOCIATIONS.put("org.apache.flink.table.runtime", "flink-table-runtime");
+ KNOWN_MODULE_ASSOCIATIONS.put("org.apache.flink.formats.raw", "flink-table-runtime");
+
+ KNOWN_MODULE_ASSOCIATIONS.put("org.codehaus.janino", "flink-table-runtime");
+ KNOWN_MODULE_ASSOCIATIONS.put("org.codehaus.commons", "flink-table-runtime");
+ KNOWN_MODULE_ASSOCIATIONS.put(
+ "org.apache.flink.table.shaded.com.jayway", "flink-table-runtime");
+ }
+
+ private final PlannerComponentClassLoader submoduleClassLoader;
+
+ private PlannerModule() {
+ try {
+ final ClassLoader flinkClassLoader = PlannerModule.class.getClassLoader();
+
+ final Path tmpDirectory =
+ Paths.get(ConfigurationUtils.parseTempDirectories(new Configuration())[0]);
+ Files.createDirectories(FileUtils.getTargetPathIfContainsSymbolicPath(tmpDirectory));
+ final Path tempFile =
+ Files.createFile(
+ tmpDirectory.resolve(
+ "flink-table-planner_" + UUID.randomUUID() + ".jar"));
+
+ final InputStream resourceStream =
+ flinkClassLoader.getResourceAsStream(FLINK_TABLE_PLANNER_FAT_JAR);
+ InputStream glutenStream =
+ flinkClassLoader.getResourceAsStream("gluten-flink-planner.jar");
+ if (resourceStream == null || glutenStream == null) {
+ throw new TableException(
+ String.format(
+ "Flink Table planner could not be found. If this happened while running a test in the IDE, "
+ + "run '%s' on the command-line, "
+ + "or add a test dependency on the flink-table-planner-loader test-jar.",
+ HINT_USAGE));
+ }
+ final Path glutenFile =
+ Files.createFile(
+ tmpDirectory.resolve(
+ "gluten-flink-planner_" + UUID.randomUUID() + ".jar"));
+
+ IOUtils.copyBytes(resourceStream, Files.newOutputStream(tempFile));
+ IOUtils.copyBytes(glutenStream, Files.newOutputStream(glutenFile));
+ tempFile.toFile().deleteOnExit();
+ glutenFile.toFile().deleteOnExit();
+
+ this.submoduleClassLoader =
+ new PlannerComponentClassLoader(
+ new URL[] {
+ glutenFile.toUri().toURL(),
+ tempFile.toUri().toURL()},
+ flinkClassLoader,
+ OWNER_CLASSPATH,
+ COMPONENT_CLASSPATH,
+ KNOWN_MODULE_ASSOCIATIONS);
+ } catch (IOException e) {
+ throw new TableException(
+ "Could not initialize the table planner components loader.", e);
+ }
+ }
+
+ public void addUrlToClassLoader(URL url) {
+ // add the url to component url
+ this.submoduleClassLoader.addURL(url);
+ }
+
+ // Singleton lazy initialization
+
+ private static class PlannerComponentsHolder {
+ private static final PlannerModule INSTANCE = new PlannerModule();
+ }
+
+ public static PlannerModule getInstance() {
+ return PlannerComponentsHolder.INSTANCE;
+ }
+
+ // load methods for various components provided by the planner
+
+ public ExecutorFactory loadExecutorFactory() {
+ return FactoryUtil.discoverFactory(
+ this.submoduleClassLoader,
+ ExecutorFactory.class,
+ ExecutorFactory.DEFAULT_IDENTIFIER);
+ }
+
+ public PlannerFactory loadPlannerFactory() {
+ return FactoryUtil.discoverFactory(
+ this.submoduleClassLoader, PlannerFactory.class, PlannerFactory.DEFAULT_IDENTIFIER);
+ }
+
+ /**
+ * A class loader extending {@link ComponentClassLoader} which overwrites method{@link #addURL}
+ * to enable it can add url to component classloader.
+ */
+ private static class PlannerComponentClassLoader extends ComponentClassLoader {
+
+ public PlannerComponentClassLoader(
+ URL[] classpath,
+ ClassLoader ownerClassLoader,
+ String[] ownerFirstPackages,
+ String[] componentFirstPackages,
+ Map<String, String> knownPackagePrefixesModuleAssociation) {
+ super(
+ classpath,
+ ownerClassLoader,
+ ownerFirstPackages,
+ componentFirstPackages,
+ knownPackagePrefixesModuleAssociation);
+ }
+
+ @Override
+ public void addURL(URL url) {
+ super.addURL(url);
+ }
+ }
+}
diff --git a/gluten-flink/planner/pom.xml b/gluten-flink/planner/pom.xml
new file mode 100644
index 0000000000..b3fc899fe5
--- /dev/null
+++ b/gluten-flink/planner/pom.xml
@@ -0,0 +1,68 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.gluten</groupId>
+ <artifactId>gluten-flink</artifactId>
+ <version>1.4.0-SNAPSHOT</version>
+ <relativePath>../pom.xml</relativePath>
+ </parent>
+
+ <artifactId>gluten-flink-planner</artifactId>
+ <name>Gluten Flink Planner</name>
+ <packaging>jar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-api-java</artifactId>
+ <version>${flink.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-planner_${scala.binary.version}</artifactId>
+ <version>${flink.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.gluten</groupId>
+ <artifactId>gluten-flink-runtime</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-clients</artifactId>
+ <version>${flink.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>io.github.zhztheplayer</groupId>
+ <artifactId>velox4j</artifactId>
+ <version>${velox4j.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ </plugins>
+ </build>
+</project>
diff --git a/gluten-flink/planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecCalc.java b/gluten-flink/planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecCalc.java
new file mode 100644
index 0000000000..18d63c5041
--- /dev/null
+++ b/gluten-flink/planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecCalc.java
@@ -0,0 +1,152 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.nodes.exec.stream;
+
+import org.apache.gluten.rexnode.Utils;
+import org.apache.gluten.util.LogicalTypeConverter;
+import org.apache.gluten.rexnode.RexNodeConverter;
+import org.apache.gluten.table.runtime.operators.GlutenCalOperator;
+
+import io.github.zhztheplayer.velox4j.connector.ExternalStreamTableHandle;
+import io.github.zhztheplayer.velox4j.expression.TypedExpr;
+import io.github.zhztheplayer.velox4j.plan.FilterNode;
+import io.github.zhztheplayer.velox4j.plan.PlanNode;
+import io.github.zhztheplayer.velox4j.plan.ProjectNode;
+import io.github.zhztheplayer.velox4j.plan.TableScanNode;
+import io.github.zhztheplayer.velox4j.serde.Serde;
+import io.github.zhztheplayer.velox4j.type.Type;
+import org.apache.calcite.rex.RexNode;
+import org.apache.flink.FlinkVersion;
+import org.apache.flink.api.dag.Transformation;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.planner.delegation.PlannerBase;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNode;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeMetadata;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecEdge;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeConfig;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeContext;
+import org.apache.flink.table.planner.plan.nodes.exec.InputProperty;
+import org.apache.flink.table.planner.plan.nodes.exec.common.CommonExecCalc;
+import org.apache.flink.table.planner.plan.nodes.exec.utils.ExecNodeUtil;
+import org.apache.flink.table.planner.plan.nodes.exec.utils.TransformationMetadata;
+import org.apache.flink.table.runtime.operators.TableStreamOperator;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+import javax.annotation.Nullable;
+
+import java.util.Collections;
+import java.util.List;
+
+/** Gluten Stream {@link ExecNode} for Calc to use {@link GlutenCalOperator}. */
+@ExecNodeMetadata(
+ name = "stream-exec-calc",
+ version = 1,
+ producedTransformations = CommonExecCalc.CALC_TRANSFORMATION,
+ minPlanVersion = FlinkVersion.v1_15,
+ minStateVersion = FlinkVersion.v1_15)
+public class StreamExecCalc extends CommonExecCalc implements StreamExecNode<RowData> {
+
+ public StreamExecCalc(
+ ReadableConfig tableConfig,
+ List<RexNode> projection,
+ @Nullable RexNode condition,
+ InputProperty inputProperty,
+ RowType outputType,
+ String description) {
+ this(
+ ExecNodeContext.newNodeId(),
+ ExecNodeContext.newContext(StreamExecCalc.class),
+ ExecNodeContext.newPersistedConfig(StreamExecCalc.class, tableConfig),
+ projection,
+ condition,
+ Collections.singletonList(inputProperty),
+ outputType,
+ description);
+ }
+
+ @JsonCreator
+ public StreamExecCalc(
+ @JsonProperty(FIELD_NAME_ID) int id,
+ @JsonProperty(FIELD_NAME_TYPE) ExecNodeContext context,
+ @JsonProperty(FIELD_NAME_CONFIGURATION) ReadableConfig persistedConfig,
+ @JsonProperty(FIELD_NAME_PROJECTION) List<RexNode> projection,
+ @JsonProperty(FIELD_NAME_CONDITION) @Nullable RexNode condition,
+ @JsonProperty(FIELD_NAME_INPUT_PROPERTIES) List<InputProperty> inputProperties,
+ @JsonProperty(FIELD_NAME_OUTPUT_TYPE) RowType outputType,
+ @JsonProperty(FIELD_NAME_DESCRIPTION) String description) {
+ super(
+ id,
+ context,
+ persistedConfig,
+ projection,
+ condition,
+ TableStreamOperator.class,
+ true, // retainHeader
+ inputProperties,
+ outputType,
+ description);
+ }
+
+ @Override
+ public Transformation<RowData> translateToPlanInternal(
+ PlannerBase planner, ExecNodeConfig config) {
+ final ExecEdge inputEdge = getInputEdges().get(0);
+ final Transformation<RowData> inputTransform =
+ (Transformation<RowData>) inputEdge.translateToPlan(planner);
+
+ Type inputType = LogicalTypeConverter.toVLType(inputEdge.getOutputType());
+ List<String> inNames = Utils.getNamesFromRowType(inputEdge.getOutputType());
+ // add a mock input as velox not allow the source is empty.
+ // TODO: remove mock table scan.
+ PlanNode mockInput = new TableScanNode(
+ String.valueOf(ExecNodeContext.newNodeId()),
+ inputType,
+ new ExternalStreamTableHandle("connector-external-stream"),
+ List.of());
+ PlanNode filter = null;
+ if (condition != null) {
+ filter = new FilterNode(
+ String.valueOf(getId()),
+ List.of(mockInput),
+ RexNodeConverter.toTypedExpr(condition, inNames));
+ }
+ List<TypedExpr> projectExprs = RexNodeConverter.toTypedExpr(projection, inNames);
+ PlanNode project = new ProjectNode(
+ String.valueOf(ExecNodeContext.newNodeId()),
+ filter == null ? List.of() : List.of(filter),
+ Utils.getNamesFromRowType(getOutputType()),
+ projectExprs);
+ // TODO: velo4j not support serializable now.
+ Utils.registerRegistry();
+ String plan = Serde.toJson(project);
+ String inputStr = Serde.toJson(inputType);
+ Type outputType = LogicalTypeConverter.toVLType(getOutputType());
+ String outputStr = Serde.toJson(outputType);
+ final GlutenCalOperator calOperator = new GlutenCalOperator(plan, mockInput.getId(), inputStr, outputStr);
+ return ExecNodeUtil.createOneInputTransformation(
+ inputTransform,
+ new TransformationMetadata("gluten-calc", "Gluten cal operator"),
+ calOperator,
+ InternalTypeInfo.of(getOutputType()),
+ inputTransform.getParallelism(),
+ false);
+ }
+}
diff --git a/gluten-flink/planner/src/main/java/org/apache/gluten/rexnode/FunctionMappings.java b/gluten-flink/planner/src/main/java/org/apache/gluten/rexnode/FunctionMappings.java
new file mode 100644
index 0000000000..f299491201
--- /dev/null
+++ b/gluten-flink/planner/src/main/java/org/apache/gluten/rexnode/FunctionMappings.java
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.gluten.rexnode;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/** Mapping of flink function and velox function. */
+public class FunctionMappings {
+ // A map stores the relationship between flink function name and velox function.
+ private static Map<String, String> functionMappings = new HashMap() {
+ {
+ // TODO: support more functions.
+ put(">", "greaterthan");
+ put("<", "lessthan");
+ put("=", "equalto");
+ put("CAST", "cast");
+ put("CASE", "case");
+ put("*", "multiply");
+ }
+ };
+
+ public static String toVeloxFunction(String funcName) {
+ if (functionMappings.containsKey(funcName)) {
+ return functionMappings.get(funcName);
+ } else {
+ throw new RuntimeException("Function not supported: " + funcName);
+ }
+ }
+}
diff --git a/gluten-flink/planner/src/main/java/org/apache/gluten/rexnode/RexNodeConverter.java b/gluten-flink/planner/src/main/java/org/apache/gluten/rexnode/RexNodeConverter.java
new file mode 100644
index 0000000000..ad1856dfb9
--- /dev/null
+++ b/gluten-flink/planner/src/main/java/org/apache/gluten/rexnode/RexNodeConverter.java
@@ -0,0 +1,150 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.gluten.rexnode;
+
+import io.github.zhztheplayer.velox4j.expression.CallTypedExpr;
+import io.github.zhztheplayer.velox4j.expression.ConstantTypedExpr;
+import io.github.zhztheplayer.velox4j.expression.FieldAccessTypedExpr;
+import io.github.zhztheplayer.velox4j.expression.TypedExpr;
+import io.github.zhztheplayer.velox4j.type.BigIntType;
+import io.github.zhztheplayer.velox4j.type.BooleanType;
+import io.github.zhztheplayer.velox4j.type.DecimalType;
+import io.github.zhztheplayer.velox4j.type.IntegerType;
+import io.github.zhztheplayer.velox4j.type.RowType;
+import io.github.zhztheplayer.velox4j.type.TimestampType;
+import io.github.zhztheplayer.velox4j.type.Type;
+import io.github.zhztheplayer.velox4j.type.VarCharType;
+import io.github.zhztheplayer.velox4j.variant.BigIntValue;
+import io.github.zhztheplayer.velox4j.variant.BooleanValue;
+import io.github.zhztheplayer.velox4j.variant.DoubleValue;
+import io.github.zhztheplayer.velox4j.variant.HugeIntValue;
+import io.github.zhztheplayer.velox4j.variant.IntegerValue;
+import io.github.zhztheplayer.velox4j.variant.SmallIntValue;
+import io.github.zhztheplayer.velox4j.variant.TinyIntValue;
+import io.github.zhztheplayer.velox4j.variant.VarBinaryValue;
+import io.github.zhztheplayer.velox4j.variant.VarCharValue;
+import io.github.zhztheplayer.velox4j.variant.Variant;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexFieldAccess;
+import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexLiteral;
+import org.apache.calcite.rex.RexNode;
+
+import java.math.BigDecimal;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/** Convertor to convert RexNode to velox TypedExpr */
+public class RexNodeConverter {
+
+ public static TypedExpr toTypedExpr(RexNode rexNode, List<String> inNames) {
+ if (rexNode instanceof RexLiteral) {
+ RexLiteral literal = (RexLiteral) rexNode;
+ return new ConstantTypedExpr(
+ toType(literal.getType()),
+ toVariant(literal),
+ null);
+ } else if (rexNode instanceof RexCall) {
+ RexCall rexCall = (RexCall) rexNode;
+ List<TypedExpr> params = toTypedExpr(rexCall.getOperands(), inNames);
+ Type nodeType = toType(rexCall.getType());
+ return new CallTypedExpr(
+ nodeType,
+ params,
+ FunctionMappings.toVeloxFunction(rexCall.getOperator().getName()));
+ } else if (rexNode instanceof RexInputRef) {
+ RexInputRef inputRef = (RexInputRef) rexNode;
+ return FieldAccessTypedExpr.create(
+ toType(inputRef.getType()),
+ inNames.get(inputRef.getIndex()));
+ } else if (rexNode instanceof RexFieldAccess) {
+ RexFieldAccess fieldAccess = (RexFieldAccess) rexNode;
+ return FieldAccessTypedExpr.create(
+ toType(fieldAccess.getType()),
+ fieldAccess.getField().getName());
+ } else {
+ throw new RuntimeException("Unrecognized RexNode: " + rexNode.getClass().getName());
+ }
+ }
+
+ public static List<TypedExpr> toTypedExpr(List<RexNode> rexNodes, List<String> inNames) {
+ return rexNodes.stream()
+ .map(rexNode -> toTypedExpr(rexNode, inNames))
+ .collect(Collectors.toList());
+ }
+
+ // TODO: use LogicalRelDataTypeConverter
+ public static Type toType(RelDataType relDataType) {
+ switch (relDataType.getSqlTypeName()) {
+ case BOOLEAN:
+ return new BooleanType();
+ case INTEGER:
+ return new IntegerType();
+ case BIGINT:
+ return new BigIntType();
+ case VARCHAR:
+ return new VarCharType();
+ case ROW:
+ List<Type> children = relDataType.getFieldList().stream()
+ .map(
+ field ->
+ toType(field.getType())
+ ).collect(Collectors.toList());
+ return new RowType(relDataType.getFieldNames(), children);
+ case TIMESTAMP:
+ return new TimestampType();
+ case DECIMAL:
+ return new DecimalType(relDataType.getPrecision(), relDataType.getScale());
+ default:
+ throw new RuntimeException("Unsupported type: " + relDataType.getSqlTypeName());
+ }
+ }
+
+ public static Variant toVariant(RexLiteral literal) {
+ switch (literal.getType().getSqlTypeName()) {
+ case BOOLEAN:
+ return new BooleanValue((boolean) literal.getValue());
+ case TINYINT:
+ return new TinyIntValue(Integer.valueOf(literal.getValue().toString()));
+ case SMALLINT:
+ return new SmallIntValue(Integer.valueOf(literal.getValue().toString()));
+ case INTEGER:
+ return new IntegerValue(Integer.valueOf(literal.getValue().toString()));
+ case BIGINT:
+ return new BigIntValue(Long.valueOf(literal.getValue().toString()));
+ case DOUBLE:
+ return new DoubleValue(Double.valueOf(literal.getValue().toString()));
+ case VARCHAR:
+ return new VarCharValue(literal.getValue().toString());
+ case BINARY:
+ return new VarBinaryValue(literal.getValue().toString());
+ case DECIMAL:
+ // TODO: fix precision check
+ BigDecimal bigDecimal = literal.getValueAs(BigDecimal.class);
+ if (bigDecimal.precision() <= 18) {
+ return new BigIntValue(bigDecimal.unscaledValue().longValueExact());
+ } else {
+ return new HugeIntValue(bigDecimal.unscaledValue());
+ }
+ default:
+ throw new RuntimeException(
+ "Unsupported rex node type: " + literal.getType().getSqlTypeName());
+ }
+ }
+
+}
diff --git a/gluten-flink/planner/src/main/java/org/apache/gluten/rexnode/Utils.java b/gluten-flink/planner/src/main/java/org/apache/gluten/rexnode/Utils.java
new file mode 100644
index 0000000000..6072b7b85e
--- /dev/null
+++ b/gluten-flink/planner/src/main/java/org/apache/gluten/rexnode/Utils.java
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.gluten.rexnode;
+
+import io.github.zhztheplayer.velox4j.serializable.ISerializableRegistry;
+import io.github.zhztheplayer.velox4j.variant.VariantRegistry;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+
+import java.util.List;
+
+/** Utility to store some useful functions. */
+public class Utils {
+
+ private static boolean registryInitialized = false;
+
+ // Get names for project node.
+ public static List<String> getNamesFromRowType(LogicalType logicalType) {
+ if (logicalType instanceof RowType) {
+ RowType rowType = (RowType) logicalType;
+ return rowType.getFieldNames();
+ } else {
+ throw new RuntimeException("Output type is not row type: " + logicalType);
+ }
+ }
+
+ // Init serialize related registries.
+ public static void registerRegistry() {
+ if (!registryInitialized) {
+ registryInitialized = true;
+ VariantRegistry.registerAll();
+ ISerializableRegistry.registerAll();
+ }
+ }
+}
diff --git a/gluten-flink/pom.xml b/gluten-flink/pom.xml
new file mode 100644
index 0000000000..696d16ace5
--- /dev/null
+++ b/gluten-flink/pom.xml
@@ -0,0 +1,41 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.gluten</groupId>
+ <artifactId>gluten-parent</artifactId>
+ <version>1.4.0-SNAPSHOT</version>
+ <relativePath>../pom.xml</relativePath>
+ </parent>
+
+ <artifactId>gluten-flink</artifactId>
+ <name>Gluten Flink</name>
+ <packaging>pom</packaging>
+
+ <modules>
+ <module>planner</module>
+ <module>loader</module>
+ <module>runtime</module>
+ </modules>
+
+ <properties>
+ <flink.version>1.19.2</flink.version>
+ <velox4j.version>0.1.0-SNAPSHOT</velox4j.version>
+ </properties>
+ </project>
diff --git a/gluten-flink/runtime/pom.xml b/gluten-flink/runtime/pom.xml
new file mode 100644
index 0000000000..b02dafbefa
--- /dev/null
+++ b/gluten-flink/runtime/pom.xml
@@ -0,0 +1,69 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.gluten</groupId>
+ <artifactId>gluten-flink</artifactId>
+ <version>1.4.0-SNAPSHOT</version>
+ <relativePath>../pom.xml</relativePath>
+ </parent>
+
+ <artifactId>gluten-flink-runtime</artifactId>
+ <name>Gluten Flink Runtime</name>
+ <packaging>jar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-streaming-java</artifactId>
+ <version>${flink.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-runtime</artifactId>
+ <version>${flink.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>io.github.zhztheplayer</groupId>
+ <artifactId>velox4j</artifactId>
+ <version>${velox4j.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.arrow</groupId>
+ <artifactId>arrow-memory-core</artifactId>
+ <version>18.1.0</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-clients</artifactId>
+ <version>${flink.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+
+ </plugins>
+ </build>
+</project>
diff --git a/gluten-flink/runtime/src/main/java/org/apache/flink/client/StreamGraphTranslator.java b/gluten-flink/runtime/src/main/java/org/apache/flink/client/StreamGraphTranslator.java
new file mode 100644
index 0000000000..f2ce9ca804
--- /dev/null
+++ b/gluten-flink/runtime/src/main/java/org/apache/flink/client/StreamGraphTranslator.java
@@ -0,0 +1,168 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.client;
+
+import org.apache.flink.api.dag.Pipeline;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.jobgraph.JobVertex;
+import org.apache.flink.streaming.api.graph.StreamConfig;
+import org.apache.flink.streaming.api.graph.StreamEdge;
+import org.apache.flink.streaming.api.graph.StreamGraph;
+import org.apache.flink.streaming.api.operators.SimpleOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
+import org.apache.gluten.streaming.api.operators.GlutenOneInputOperatorFactory;
+import org.apache.gluten.table.runtime.operators.GlutenSourceFunction;
+
+import io.github.zhztheplayer.velox4j.plan.PlanNode;
+import org.apache.gluten.streaming.api.operators.GlutenOperator;
+import org.apache.gluten.streaming.api.operators.GlutenStreamSource;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+
+/**
+ * {@link FlinkPipelineTranslator} for DataStream API {@link StreamGraph StreamGraphs}.
+ *
+ * <p>Note: this is used through reflection in {@link
+ * org.apache.flink.client.FlinkPipelineTranslationUtil}.
+ */
+@SuppressWarnings("unused")
+public class StreamGraphTranslator implements FlinkPipelineTranslator {
+
+ private static final Logger LOG = LoggerFactory.getLogger(StreamGraphTranslator.class);
+
+ private final ClassLoader userClassloader;
+
+ public StreamGraphTranslator(ClassLoader userClassloader) {
+ this.userClassloader = userClassloader;
+ }
+
+ @Override
+ public JobGraph translateToJobGraph(
+ Pipeline pipeline, Configuration optimizerConfiguration, int defaultParallelism) {
+ checkArgument(
+ pipeline instanceof StreamGraph, "Given pipeline is not a DataStream StreamGraph.");
+
+ StreamGraph streamGraph = (StreamGraph) pipeline;
+ JobGraph jobGraph = streamGraph.getJobGraph(userClassloader, null);
+ return mergeGlutenOperators(jobGraph);
+ }
+
+ @Override
+ public String translateToJSONExecutionPlan(Pipeline pipeline) {
+ checkArgument(
+ pipeline instanceof StreamGraph, "Given pipeline is not a DataStream StreamGraph.");
+
+ StreamGraph streamGraph = (StreamGraph) pipeline;
+
+ return streamGraph.getStreamingPlanAsJSON();
+ }
+
+ @Override
+ public boolean canTranslate(Pipeline pipeline) {
+ return pipeline instanceof StreamGraph;
+ }
+
+ private JobGraph mergeGlutenOperators(JobGraph jobGraph) {
+ for (JobVertex vertex : jobGraph.getVertices()) {
+ StreamConfig streamConfig = new StreamConfig(vertex.getConfiguration());
+ buildGlutenChains(streamConfig);
+ }
+ return jobGraph;
+ }
+
+ private void buildGlutenChains(
+ StreamConfig taskConfig) {
+ // TODO: only support head operator now.
+ if (isGlutenOperator(taskConfig)) {
+ while (true) {
+ List<StreamEdge> outEdges = taskConfig.getChainedOutputs(userClassloader);
+ if (outEdges.size() != 1) {
+ // only support operators have one output.
+ break;
+ }
+ StreamEdge outEdge = outEdges.get(0);
+ Map<Integer, StreamConfig> chainedTasks =
+ taskConfig.getTransitiveChainedTaskConfigs(userClassloader);
+ StreamConfig outTask = chainedTasks.get(outEdge.getTargetId());
+ if (outTask != null) {
+ if (isGlutenOperator(outTask)) {
+ GlutenOperator outOperator = getGlutenOperator(outTask);
+ PlanNode outNode = outOperator.getPlanNode();
+ GlutenOperator sourceOperator = getGlutenOperator(taskConfig);
+ if (outNode != null) {
+ outNode.setSources(List.of(sourceOperator.getPlanNode()));
+ } else {
+ outNode = sourceOperator.getPlanNode();
+ }
+ if (sourceOperator instanceof GlutenStreamSource) {
+ taskConfig.setStreamOperator(
+ new GlutenStreamSource(
+ new GlutenSourceFunction(
+ outNode,
+ outOperator.getOutputType(),
+ sourceOperator.getId())));
+ } else {
+ // TODO: support GlutenCombinedOperator.
+ //taskConfig.setStreamOperator(
+ // new GlutenCombinedOperator(outNode));
+ }
+ // TODO: stream edges need to be reset.
+ taskConfig.setTransitiveChainedTaskConfigs(
+ outTask.getTransitiveChainedTaskConfigs(userClassloader));
+ taskConfig.setChainedOutputs(outTask.getChainedOutputs(userClassloader));
+ taskConfig.setOperatorNonChainedOutputs(
+ outTask.getOperatorNonChainedOutputs(userClassloader));
+ taskConfig.serializeAllConfigs();
+ } else {
+ break;
+ }
+ } else {
+ break;
+ }
+ }
+ }
+ }
+
+ private boolean isGlutenOperator(StreamConfig taskConfig) {
+ StreamOperatorFactory operatorFactory = taskConfig.getStreamOperatorFactory(userClassloader);
+ if (operatorFactory instanceof SimpleOperatorFactory) {
+ return taskConfig.getStreamOperator(userClassloader) instanceof GlutenOperator;
+ } else if (operatorFactory instanceof GlutenOneInputOperatorFactory) {
+ return true;
+ }
+ return false;
+ }
+
+ private GlutenOperator getGlutenOperator(StreamConfig taskConfig) {
+ StreamOperatorFactory operatorFactory = taskConfig.getStreamOperatorFactory(userClassloader);
+ if (operatorFactory instanceof SimpleOperatorFactory) {
+ return taskConfig.getStreamOperator(userClassloader);
+ } else if (operatorFactory instanceof GlutenOneInputOperatorFactory) {
+ return ((GlutenOneInputOperatorFactory) operatorFactory).getOperator();
+ }
+ return null;
+ }
+}
diff --git a/gluten-flink/runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java b/gluten-flink/runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
new file mode 100644
index 0000000000..101ec77371
--- /dev/null
+++ b/gluten-flink/runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
@@ -0,0 +1,823 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.taskexecutor;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.JMXServerOptions;
+import org.apache.flink.configuration.RpcOptions;
+import org.apache.flink.configuration.TaskManagerOptions;
+import org.apache.flink.configuration.TaskManagerOptionsInternal;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.plugin.PluginManager;
+import org.apache.flink.core.plugin.PluginUtils;
+import org.apache.flink.core.security.FlinkSecurityManager;
+import org.apache.flink.management.jmx.JMXService;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.runtime.blob.BlobCacheService;
+import org.apache.flink.runtime.blob.BlobUtils;
+import org.apache.flink.runtime.blob.TaskExecutorBlobService;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.entrypoint.ClusterEntrypointUtils;
+import org.apache.flink.runtime.entrypoint.DeterminismEnvelope;
+import org.apache.flink.runtime.entrypoint.FlinkParseException;
+import org.apache.flink.runtime.entrypoint.WorkingDirectory;
+import org.apache.flink.runtime.externalresource.ExternalResourceInfoProvider;
+import org.apache.flink.runtime.externalresource.ExternalResourceUtils;
+import org.apache.flink.runtime.heartbeat.HeartbeatServices;
+import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
+import org.apache.flink.runtime.highavailability.HighAvailabilityServicesUtils;
+import org.apache.flink.runtime.io.network.partition.TaskExecutorPartitionTrackerImpl;
+import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalException;
+import org.apache.flink.runtime.metrics.MetricRegistry;
+import org.apache.flink.runtime.metrics.MetricRegistryConfiguration;
+import org.apache.flink.runtime.metrics.MetricRegistryImpl;
+import org.apache.flink.runtime.metrics.ReporterSetup;
+import org.apache.flink.runtime.metrics.TraceReporterSetup;
+import org.apache.flink.runtime.metrics.groups.TaskManagerMetricGroup;
+import org.apache.flink.runtime.metrics.util.MetricUtils;
+import org.apache.flink.runtime.rpc.AddressResolution;
+import org.apache.flink.runtime.rpc.FatalErrorHandler;
+import org.apache.flink.runtime.rpc.RpcService;
+import org.apache.flink.runtime.rpc.RpcSystem;
+import org.apache.flink.runtime.rpc.RpcSystemUtils;
+import org.apache.flink.runtime.rpc.RpcUtils;
+import org.apache.flink.runtime.security.SecurityConfiguration;
+import org.apache.flink.runtime.security.SecurityUtils;
+import org.apache.flink.runtime.security.token.DelegationTokenReceiverRepository;
+import org.apache.flink.runtime.state.changelog.StateChangelogStorageLoader;
+import org.apache.flink.runtime.taskmanager.MemoryLogger;
+import org.apache.flink.runtime.util.ConfigurationParserUtils;
+import org.apache.flink.runtime.util.EnvironmentInformation;
+import org.apache.flink.runtime.util.Hardware;
+import org.apache.flink.runtime.util.JvmShutdownSafeguard;
+import org.apache.flink.runtime.util.LeaderRetrievalUtils;
+import org.apache.flink.runtime.util.SignalHandler;
+import org.apache.flink.util.AbstractID;
+import org.apache.flink.util.AutoCloseableAsync;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.ExecutorUtils;
+import org.apache.flink.util.FlinkException;
+import org.apache.flink.util.Reference;
+import org.apache.flink.util.ShutdownHookUtil;
+import org.apache.flink.util.StringUtils;
+import org.apache.flink.util.TaskManagerExceptionUtils;
+import org.apache.flink.util.concurrent.ExecutorThreadFactory;
+import org.apache.flink.util.concurrent.FutureUtils;
+import org.apache.flink.util.function.FunctionUtils;
+
+import io.github.zhztheplayer.velox4j.Velox4j;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.concurrent.GuardedBy;
+
+import java.io.IOException;
+import java.lang.reflect.UndeclaredThrowableException;
+import java.net.InetAddress;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * This class is the executable entry point for the task manager in yarn or standalone mode. It
+ * constructs the related components (network, I/O manager, memory manager, RPC service, HA service)
+ * and starts them.
+ */
+public class TaskManagerRunner implements FatalErrorHandler {
+
+ private static final Logger LOG = LoggerFactory.getLogger(TaskManagerRunner.class);
+
+ private static final long FATAL_ERROR_SHUTDOWN_TIMEOUT_MS = 10000L;
+
+ private static final int SUCCESS_EXIT_CODE = 0;
+ @VisibleForTesting public static final int FAILURE_EXIT_CODE = 1;
+
+ private final Thread shutdownHook;
+
+ private final Object lock = new Object();
+
+ private final Configuration configuration;
+
+ private final Time timeout;
+
+ private final PluginManager pluginManager;
+
+ private final TaskExecutorServiceFactory taskExecutorServiceFactory;
+
+ private final CompletableFuture<Result> terminationFuture;
+
+ @GuardedBy("lock")
+ private DeterminismEnvelope<ResourceID> resourceId;
+
+ /** Executor used to run future callbacks. */
+ @GuardedBy("lock")
+ private ExecutorService executor;
+
+ @GuardedBy("lock")
+ private RpcSystem rpcSystem;
+
+ @GuardedBy("lock")
+ private RpcService rpcService;
+
+ @GuardedBy("lock")
+ private HighAvailabilityServices highAvailabilityServices;
+
+ @GuardedBy("lock")
+ private MetricRegistryImpl metricRegistry;
+
+ @GuardedBy("lock")
+ private BlobCacheService blobCacheService;
+
+ @GuardedBy("lock")
+ private DeterminismEnvelope<WorkingDirectory> workingDirectory;
+
+ @GuardedBy("lock")
+ private TaskExecutorService taskExecutorService;
+
+ @GuardedBy("lock")
+ private boolean shutdown;
+
+ public TaskManagerRunner(
+ Configuration configuration,
+ PluginManager pluginManager,
+ TaskExecutorServiceFactory taskExecutorServiceFactory)
+ throws Exception {
+ this.configuration = checkNotNull(configuration);
+ this.pluginManager = checkNotNull(pluginManager);
+ this.taskExecutorServiceFactory = checkNotNull(taskExecutorServiceFactory);
+
+ timeout = Time.fromDuration(configuration.get(RpcOptions.ASK_TIMEOUT_DURATION));
+
+ this.terminationFuture = new CompletableFuture<>();
+ this.shutdown = false;
+
+ this.shutdownHook =
+ ShutdownHookUtil.addShutdownHook(
+ () -> this.closeAsync(Result.JVM_SHUTDOWN).join(),
+ getClass().getSimpleName(),
+ LOG);
+ }
+
+ private void startTaskManagerRunnerServices() throws Exception {
+ synchronized (lock) {
+ rpcSystem = RpcSystem.load(configuration);
+
+ this.executor =
+ Executors.newScheduledThreadPool(
+ Hardware.getNumberCPUCores(),
+ new ExecutorThreadFactory("taskmanager-future"));
+
+ highAvailabilityServices =
+ HighAvailabilityServicesUtils.createHighAvailabilityServices(
+ configuration,
+ executor,
+ AddressResolution.NO_ADDRESS_RESOLUTION,
+ rpcSystem,
+ this);
+
+ JMXService.startInstance(configuration.get(JMXServerOptions.JMX_SERVER_PORT));
+
+ rpcService = createRpcService(configuration, highAvailabilityServices, rpcSystem);
+
+ this.resourceId =
+ getTaskManagerResourceID(
+ configuration, rpcService.getAddress(), rpcService.getPort());
+
+ this.workingDirectory =
+ ClusterEntrypointUtils.createTaskManagerWorkingDirectory(
+ configuration, resourceId);
+
+ LOG.info("Using working directory: {}", workingDirectory);
+
+ HeartbeatServices heartbeatServices =
+ HeartbeatServices.fromConfiguration(configuration);
+
+ metricRegistry =
+ new MetricRegistryImpl(
+ MetricRegistryConfiguration.fromConfiguration(
+ configuration,
+ rpcSystem.getMaximumMessageSizeInBytes(configuration)),
+ ReporterSetup.fromConfiguration(configuration, pluginManager),
+ TraceReporterSetup.fromConfiguration(configuration, pluginManager));
+
+ final RpcService metricQueryServiceRpcService =
+ MetricUtils.startRemoteMetricsRpcService(
+ configuration,
+ rpcService.getAddress(),
+ configuration.get(TaskManagerOptions.BIND_HOST),
+ rpcSystem);
+ metricRegistry.startQueryService(metricQueryServiceRpcService, resourceId.unwrap());
+
+ blobCacheService =
+ BlobUtils.createBlobCacheService(
+ configuration,
+ Reference.borrowed(workingDirectory.unwrap().getBlobStorageDirectory()),
+ highAvailabilityServices.createBlobStore(),
+ null);
+
+ final ExternalResourceInfoProvider externalResourceInfoProvider =
+ ExternalResourceUtils.createStaticExternalResourceInfoProviderFromConfig(
+ configuration, pluginManager);
+
+ final DelegationTokenReceiverRepository delegationTokenReceiverRepository =
+ new DelegationTokenReceiverRepository(configuration, pluginManager);
+
+ taskExecutorService =
+ taskExecutorServiceFactory.createTaskExecutor(
+ this.configuration,
+ this.resourceId.unwrap(),
+ rpcService,
+ highAvailabilityServices,
+ heartbeatServices,
+ metricRegistry,
+ blobCacheService,
+ false,
+ externalResourceInfoProvider,
+ workingDirectory.unwrap(),
+ this,
+ delegationTokenReceiverRepository);
+
+ handleUnexpectedTaskExecutorServiceTermination();
+
+ MemoryLogger.startIfConfigured(
+ LOG, configuration, terminationFuture.thenAccept(ignored -> {}));
+ }
+ }
+
+ @GuardedBy("lock")
+ private void handleUnexpectedTaskExecutorServiceTermination() {
+ taskExecutorService
+ .getTerminationFuture()
+ .whenComplete(
+ (unused, throwable) -> {
+ synchronized (lock) {
+ if (!shutdown) {
+ onFatalError(
+ new FlinkException(
+ "Unexpected termination of the TaskExecutor.",
+ throwable));
+ }
+ }
+ });
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Lifecycle management
+ // --------------------------------------------------------------------------------------------
+
+ public void start() throws Exception {
+ synchronized (lock) {
+ startTaskManagerRunnerServices();
+ taskExecutorService.start();
+ }
+ }
+
+ public void close() throws Exception {
+ try {
+ closeAsync().get();
+ } catch (ExecutionException e) {
+ ExceptionUtils.rethrowException(ExceptionUtils.stripExecutionException(e));
+ }
+ }
+
+ public CompletableFuture<Result> closeAsync() {
+ return closeAsync(Result.SUCCESS);
+ }
+
+ private CompletableFuture<Result> closeAsync(Result terminationResult) {
+ synchronized (lock) {
+ // remove shutdown hook to prevent resource leaks
+ ShutdownHookUtil.removeShutdownHook(shutdownHook, this.getClass().getSimpleName(), LOG);
+
+ if (shutdown) {
+ return terminationFuture;
+ }
+
+ final CompletableFuture<Void> taskManagerTerminationFuture;
+ if (taskExecutorService != null) {
+ taskManagerTerminationFuture = taskExecutorService.closeAsync();
+ } else {
+ taskManagerTerminationFuture = FutureUtils.completedVoidFuture();
+ }
+
+ final CompletableFuture<Void> serviceTerminationFuture =
+ FutureUtils.composeAfterwards(
+ taskManagerTerminationFuture, this::shutDownServices);
+
+ final CompletableFuture<Void> workingDirCleanupFuture =
+ FutureUtils.runAfterwards(
+ serviceTerminationFuture, () -> deleteWorkingDir(terminationResult));
+
+ final CompletableFuture<Void> rpcSystemClassLoaderCloseFuture;
+
+ if (rpcSystem != null) {
+ rpcSystemClassLoaderCloseFuture =
+ FutureUtils.runAfterwards(workingDirCleanupFuture, rpcSystem::close);
+ } else {
+ rpcSystemClassLoaderCloseFuture = FutureUtils.completedVoidFuture();
+ }
+
+ rpcSystemClassLoaderCloseFuture.whenComplete(
+ (Void ignored, Throwable throwable) -> {
+ if (throwable != null) {
+ terminationFuture.completeExceptionally(throwable);
+ } else {
+ terminationFuture.complete(terminationResult);
+ }
+ });
+
+ shutdown = true;
+ return terminationFuture;
+ }
+ }
+
+ private void deleteWorkingDir(Result terminationResult) throws IOException {
+ synchronized (lock) {
+ if (workingDirectory != null) {
+ if (!workingDirectory.isDeterministic() || terminationResult == Result.SUCCESS) {
+ workingDirectory.unwrap().delete();
+ }
+ }
+ }
+ }
+
+ private CompletableFuture<Void> shutDownServices() {
+ synchronized (lock) {
+ Collection<CompletableFuture<Void>> terminationFutures = new ArrayList<>(3);
+ Exception exception = null;
+
+ try {
+ JMXService.stopInstance();
+ } catch (Exception e) {
+ exception = ExceptionUtils.firstOrSuppressed(e, exception);
+ }
+
+ if (blobCacheService != null) {
+ try {
+ blobCacheService.close();
+ } catch (Exception e) {
+ exception = ExceptionUtils.firstOrSuppressed(e, exception);
+ }
+ }
+
+ if (metricRegistry != null) {
+ try {
+ terminationFutures.add(metricRegistry.closeAsync());
+ } catch (Exception e) {
+ exception = ExceptionUtils.firstOrSuppressed(e, exception);
+ }
+ }
+
+ if (highAvailabilityServices != null) {
+ try {
+ highAvailabilityServices.close();
+ } catch (Exception e) {
+ exception = ExceptionUtils.firstOrSuppressed(e, exception);
+ }
+ }
+
+ if (rpcService != null) {
+ terminationFutures.add(rpcService.closeAsync());
+ }
+
+ if (executor != null) {
+ terminationFutures.add(
+ ExecutorUtils.nonBlockingShutdown(
+ timeout.toMilliseconds(), TimeUnit.MILLISECONDS, executor));
+ }
+
+ if (exception != null) {
+ terminationFutures.add(FutureUtils.completedExceptionally(exception));
+ }
+
+ return FutureUtils.completeAll(terminationFutures);
+ }
+ }
+
+ // export the termination future for caller to know it is terminated
+ public CompletableFuture<Result> getTerminationFuture() {
+ return terminationFuture;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // FatalErrorHandler methods
+ // --------------------------------------------------------------------------------------------
+
+ @Override
+ public void onFatalError(Throwable exception) {
+ TaskManagerExceptionUtils.tryEnrichTaskManagerError(exception);
+ LOG.error(
+ "Fatal error occurred while executing the TaskManager. Shutting it down...",
+ exception);
+
+ if (ExceptionUtils.isJvmFatalOrOutOfMemoryError(exception)) {
+ terminateJVM();
+ } else {
+ closeAsync(Result.FAILURE);
+
+ FutureUtils.orTimeout(
+ terminationFuture,
+ FATAL_ERROR_SHUTDOWN_TIMEOUT_MS,
+ TimeUnit.MILLISECONDS,
+ String.format(
+ "Waiting for TaskManager shutting down timed out after %s ms.",
+ FATAL_ERROR_SHUTDOWN_TIMEOUT_MS));
+ }
+ }
+
+ private void terminateJVM() {
+ FlinkSecurityManager.forceProcessExit(FAILURE_EXIT_CODE);
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Static entry point
+ // --------------------------------------------------------------------------------------------
+
+ public static void main(String[] args) throws Exception {
+ // startup checks and logging
+ EnvironmentInformation.logEnvironmentInfo(LOG, "TaskManager", args);
+ SignalHandler.register(LOG);
+ JvmShutdownSafeguard.installAsShutdownHook(LOG);
+
+ long maxOpenFileHandles = EnvironmentInformation.getOpenFileHandlesLimit();
+
+ if (maxOpenFileHandles != -1L) {
+ LOG.info("Maximum number of open file descriptors is {}.", maxOpenFileHandles);
+ } else {
+ LOG.info("Cannot determine the maximum number of open file descriptors");
+ }
+
+ runTaskManagerProcessSecurely(args);
+ }
+
+ public static Configuration loadConfiguration(String[] args) throws FlinkParseException {
+ return ConfigurationParserUtils.loadCommonConfiguration(
+ args, TaskManagerRunner.class.getSimpleName());
+ }
+
+ public static int runTaskManager(Configuration configuration, PluginManager pluginManager)
+ throws Exception {
+ final TaskManagerRunner taskManagerRunner;
+
+ try {
+ taskManagerRunner =
+ new TaskManagerRunner(
+ configuration,
+ pluginManager,
+ TaskManagerRunner::createTaskExecutorService);
+ taskManagerRunner.start();
+ } catch (Exception exception) {
+ throw new FlinkException("Failed to start the TaskManagerRunner.", exception);
+ }
+
+ try {
+ return taskManagerRunner.getTerminationFuture().get().getExitCode();
+ } catch (Throwable t) {
+ throw new FlinkException(
+ "Unexpected failure during runtime of TaskManagerRunner.",
+ ExceptionUtils.stripExecutionException(t));
+ }
+ }
+
+ public static void runTaskManagerProcessSecurely(String[] args) {
+ Configuration configuration = null;
+
+ try {
+ configuration = loadConfiguration(args);
+ } catch (FlinkParseException fpe) {
+ LOG.error("Could not load the configuration.", fpe);
+ System.exit(FAILURE_EXIT_CODE);
+ }
+
+ runTaskManagerProcessSecurely(checkNotNull(configuration));
+ }
+
+ public static void runTaskManagerProcessSecurely(Configuration configuration) {
+ FlinkSecurityManager.setFromConfiguration(configuration);
+ final PluginManager pluginManager =
+ PluginUtils.createPluginManagerFromRootFolder(configuration);
+ FileSystem.initialize(configuration, pluginManager);
+
+ StateChangelogStorageLoader.initialize(pluginManager);
+ Velox4j.initialize();
+
+ int exitCode;
+ Throwable throwable = null;
+
+ ClusterEntrypointUtils.configureUncaughtExceptionHandler(configuration);
+ try {
+ SecurityUtils.install(new SecurityConfiguration(configuration));
+
+ exitCode =
+ SecurityUtils.getInstalledContext()
+ .runSecured(() -> runTaskManager(configuration, pluginManager));
+ } catch (Throwable t) {
+ throwable = ExceptionUtils.stripException(t, UndeclaredThrowableException.class);
+ exitCode = FAILURE_EXIT_CODE;
+ }
+
+ if (throwable != null) {
+ LOG.error("Terminating TaskManagerRunner with exit code {}.", exitCode, throwable);
+ } else {
+ LOG.info("Terminating TaskManagerRunner with exit code {}.", exitCode);
+ }
+
+ System.exit(exitCode);
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Static utilities
+ // --------------------------------------------------------------------------------------------
+
+ public static TaskExecutorService createTaskExecutorService(
+ Configuration configuration,
+ ResourceID resourceID,
+ RpcService rpcService,
+ HighAvailabilityServices highAvailabilityServices,
+ HeartbeatServices heartbeatServices,
+ MetricRegistry metricRegistry,
+ BlobCacheService blobCacheService,
+ boolean localCommunicationOnly,
+ ExternalResourceInfoProvider externalResourceInfoProvider,
+ WorkingDirectory workingDirectory,
+ FatalErrorHandler fatalErrorHandler,
+ DelegationTokenReceiverRepository delegationTokenReceiverRepository)
+ throws Exception {
+
+ final TaskExecutor taskExecutor =
+ startTaskManager(
+ configuration,
+ resourceID,
+ rpcService,
+ highAvailabilityServices,
+ heartbeatServices,
+ metricRegistry,
+ blobCacheService,
+ localCommunicationOnly,
+ externalResourceInfoProvider,
+ workingDirectory,
+ fatalErrorHandler,
+ delegationTokenReceiverRepository);
+
+ return TaskExecutorToServiceAdapter.createFor(taskExecutor);
+ }
+
+ public static TaskExecutor startTaskManager(
+ Configuration configuration,
+ ResourceID resourceID,
+ RpcService rpcService,
+ HighAvailabilityServices highAvailabilityServices,
+ HeartbeatServices heartbeatServices,
+ MetricRegistry metricRegistry,
+ TaskExecutorBlobService taskExecutorBlobService,
+ boolean localCommunicationOnly,
+ ExternalResourceInfoProvider externalResourceInfoProvider,
+ WorkingDirectory workingDirectory,
+ FatalErrorHandler fatalErrorHandler,
+ DelegationTokenReceiverRepository delegationTokenReceiverRepository)
+ throws Exception {
+
+ checkNotNull(configuration);
+ checkNotNull(resourceID);
+ checkNotNull(rpcService);
+ checkNotNull(highAvailabilityServices);
+
+ LOG.info("Starting TaskManager with ResourceID: {}", resourceID.getStringWithMetadata());
+
+ SystemOutRedirectionUtils.redirectSystemOutAndError(configuration);
+
+ String externalAddress = rpcService.getAddress();
+
+ final TaskExecutorResourceSpec taskExecutorResourceSpec =
+ TaskExecutorResourceUtils.resourceSpecFromConfig(configuration);
+
+ TaskManagerServicesConfiguration taskManagerServicesConfiguration =
+ TaskManagerServicesConfiguration.fromConfiguration(
+ configuration,
+ resourceID,
+ externalAddress,
+ localCommunicationOnly,
+ taskExecutorResourceSpec,
+ workingDirectory);
+
+ Tuple2<TaskManagerMetricGroup, MetricGroup> taskManagerMetricGroup =
+ MetricUtils.instantiateTaskManagerMetricGroup(
+ metricRegistry,
+ externalAddress,
+ resourceID,
+ taskManagerServicesConfiguration.getSystemResourceMetricsProbingInterval());
+
+ final ExecutorService ioExecutor =
+ Executors.newFixedThreadPool(
+ taskManagerServicesConfiguration.getNumIoThreads(),
+ new ExecutorThreadFactory("flink-taskexecutor-io"));
+
+ TaskManagerServices taskManagerServices =
+ TaskManagerServices.fromConfiguration(
+ taskManagerServicesConfiguration,
+ taskExecutorBlobService.getPermanentBlobService(),
+ taskManagerMetricGroup.f1,
+ ioExecutor,
+ rpcService.getScheduledExecutor(),
+ fatalErrorHandler,
+ workingDirectory);
+
+ MetricUtils.instantiateFlinkMemoryMetricGroup(
+ taskManagerMetricGroup.f1,
+ taskManagerServices.getTaskSlotTable(),
+ taskManagerServices::getManagedMemorySize);
+
+ TaskManagerConfiguration taskManagerConfiguration =
+ TaskManagerConfiguration.fromConfiguration(
+ configuration,
+ taskExecutorResourceSpec,
+ externalAddress,
+ workingDirectory.getTmpDirectory());
+
+ String metricQueryServiceAddress = metricRegistry.getMetricQueryServiceGatewayRpcAddress();
+
+ return new TaskExecutor(
+ rpcService,
+ taskManagerConfiguration,
+ highAvailabilityServices,
+ taskManagerServices,
+ externalResourceInfoProvider,
+ heartbeatServices,
+ taskManagerMetricGroup.f0,
+ metricQueryServiceAddress,
+ taskExecutorBlobService,
+ fatalErrorHandler,
+ new TaskExecutorPartitionTrackerImpl(taskManagerServices.getShuffleEnvironment()),
+ delegationTokenReceiverRepository);
+ }
+
+ /**
+ * Create a RPC service for the task manager.
+ *
+ * @param configuration The configuration for the TaskManager.
+ * @param haServices to use for the task manager hostname retrieval
+ */
+ @VisibleForTesting
+ static RpcService createRpcService(
+ final Configuration configuration,
+ final HighAvailabilityServices haServices,
+ final RpcSystem rpcSystem)
+ throws Exception {
+
+ checkNotNull(configuration);
+ checkNotNull(haServices);
+
+ return RpcUtils.createRemoteRpcService(
+ rpcSystem,
+ configuration,
+ determineTaskManagerBindAddress(configuration, haServices, rpcSystem),
+ configuration.get(TaskManagerOptions.RPC_PORT),
+ configuration.get(TaskManagerOptions.BIND_HOST),
+ configuration.getOptional(TaskManagerOptions.RPC_BIND_PORT));
+ }
+
+ private static String determineTaskManagerBindAddress(
+ final Configuration configuration,
+ final HighAvailabilityServices haServices,
+ RpcSystemUtils rpcSystemUtils)
+ throws Exception {
+
+ final String configuredTaskManagerHostname = configuration.get(TaskManagerOptions.HOST);
+
+ if (configuredTaskManagerHostname != null) {
+ LOG.info(
+ "Using configured hostname/address for TaskManager: {}.",
+ configuredTaskManagerHostname);
+ return configuredTaskManagerHostname;
+ } else {
+ return determineTaskManagerBindAddressByConnectingToResourceManager(
+ configuration, haServices, rpcSystemUtils);
+ }
+ }
+
+ private static String determineTaskManagerBindAddressByConnectingToResourceManager(
+ final Configuration configuration,
+ final HighAvailabilityServices haServices,
+ RpcSystemUtils rpcSystemUtils)
+ throws LeaderRetrievalException {
+
+ final Duration lookupTimeout = configuration.get(RpcOptions.LOOKUP_TIMEOUT_DURATION);
+
+ final InetAddress taskManagerAddress =
+ LeaderRetrievalUtils.findConnectingAddress(
+ haServices.getResourceManagerLeaderRetriever(),
+ lookupTimeout,
+ rpcSystemUtils);
+
+ LOG.info(
+ "TaskManager will use hostname/address '{}' ({}) for communication.",
+ taskManagerAddress.getHostName(),
+ taskManagerAddress.getHostAddress());
+
+ HostBindPolicy bindPolicy =
+ HostBindPolicy.fromString(configuration.get(TaskManagerOptions.HOST_BIND_POLICY));
+ return bindPolicy == HostBindPolicy.IP
+ ? taskManagerAddress.getHostAddress()
+ : taskManagerAddress.getHostName();
+ }
+
+ @VisibleForTesting
+ static DeterminismEnvelope<ResourceID> getTaskManagerResourceID(
+ Configuration config, String rpcAddress, int rpcPort) {
+
+ final String metadata =
+ config.get(TaskManagerOptionsInternal.TASK_MANAGER_RESOURCE_ID_METADATA, "");
+ return config.getOptional(TaskManagerOptions.TASK_MANAGER_RESOURCE_ID)
+ .map(
+ value ->
+ DeterminismEnvelope.deterministicValue(
+ new ResourceID(value, metadata)))
+ .orElseGet(
+ FunctionUtils.uncheckedSupplier(
+ () -> {
+ final String hostName =
+ InetAddress.getLocalHost().getHostName();
+ final String value =
+ StringUtils.isNullOrWhitespaceOnly(rpcAddress)
+ ? hostName
+ + "-"
+ + new AbstractID()
+ .toString()
+ .substring(0, 6)
+ : rpcAddress
+ + ":"
+ + rpcPort
+ + "-"
+ + new AbstractID()
+ .toString()
+ .substring(0, 6);
+ return DeterminismEnvelope.nondeterministicValue(
+ new ResourceID(value, metadata));
+ }));
+ }
+
+ /** Factory for {@link TaskExecutor}. */
+ public interface TaskExecutorServiceFactory {
+ TaskExecutorService createTaskExecutor(
+ Configuration configuration,
+ ResourceID resourceID,
+ RpcService rpcService,
+ HighAvailabilityServices highAvailabilityServices,
+ HeartbeatServices heartbeatServices,
+ MetricRegistry metricRegistry,
+ BlobCacheService blobCacheService,
+ boolean localCommunicationOnly,
+ ExternalResourceInfoProvider externalResourceInfoProvider,
+ WorkingDirectory workingDirectory,
+ FatalErrorHandler fatalErrorHandler,
+ DelegationTokenReceiverRepository delegationTokenReceiverRepository)
+ throws Exception;
+ }
+
+ public interface TaskExecutorService extends AutoCloseableAsync {
+ void start();
+
+ CompletableFuture<Void> getTerminationFuture();
+ }
+
+ public enum Result {
+ SUCCESS(SUCCESS_EXIT_CODE),
+ JVM_SHUTDOWN(FAILURE_EXIT_CODE),
+ FAILURE(FAILURE_EXIT_CODE);
+
+ private final int exitCode;
+
+ Result(int exitCode) {
+ this.exitCode = exitCode;
+ }
+
+ public int getExitCode() {
+ return exitCode;
+ }
+ }
+}
diff --git a/gluten-flink/runtime/src/main/java/org/apache/flink/streaming/runtime/translators/LegacySourceTransformationTranslator.java b/gluten-flink/runtime/src/main/java/org/apache/flink/streaming/runtime/translators/LegacySourceTransformationTranslator.java
new file mode 100644
index 0000000000..8abd5d8202
--- /dev/null
+++ b/gluten-flink/runtime/src/main/java/org/apache/flink/streaming/runtime/translators/LegacySourceTransformationTranslator.java
@@ -0,0 +1,127 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.runtime.translators;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.functions.Function;
+import org.apache.flink.streaming.api.functions.source.datagen.DataGeneratorSource;
+import org.apache.flink.streaming.api.graph.SimpleTransformationTranslator;
+import org.apache.flink.streaming.api.graph.StreamGraph;
+import org.apache.flink.streaming.api.graph.TransformationTranslator;
+import org.apache.flink.streaming.api.operators.InputFormatOperatorFactory;
+import org.apache.flink.streaming.api.operators.SimpleOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
+import org.apache.flink.streaming.api.transformations.LegacySourceTransformation;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+
+import io.github.zhztheplayer.velox4j.connector.FuzzerTableHandle;
+import io.github.zhztheplayer.velox4j.plan.TableScanNode;
+import io.github.zhztheplayer.velox4j.type.RowType;
+import org.apache.gluten.streaming.api.operators.GlutenStreamSource;
+import org.apache.gluten.table.runtime.operators.GlutenSourceFunction;
+import org.apache.gluten.util.LogicalTypeConverter;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Gluten {@link TransformationTranslator} for the {@link LegacySourceTransformation}.
+ *
+ * @param <OUT> The type of the elements that the {@link LegacySourceTransformation} we are
+ * translating is producing.
+ */
+@Internal
+public class LegacySourceTransformationTranslator<OUT>
+ extends SimpleTransformationTranslator<OUT, LegacySourceTransformation<OUT>> {
+
+ @Override
+ protected Collection<Integer> translateForBatchInternal(
+ final LegacySourceTransformation<OUT> transformation, final Context context) {
+ return translateInternal(transformation, context);
+ }
+
+ @Override
+ protected Collection<Integer> translateForStreamingInternal(
+ final LegacySourceTransformation<OUT> transformation, final Context context) {
+ return translateInternal(transformation, context);
+ }
+
+ private Collection<Integer> translateInternal(
+ final LegacySourceTransformation<OUT> transformation, final Context context) {
+ checkNotNull(transformation);
+ checkNotNull(context);
+
+ final StreamGraph streamGraph = context.getStreamGraph();
+ final String slotSharingGroup = context.getSlotSharingGroup();
+ final int transformationId = transformation.getId();
+ final ExecutionConfig executionConfig = streamGraph.getExecutionConfig();
+
+ Function userFunction = transformation.getOperator().getUserFunction();
+ StreamOperatorFactory<OUT> operatorFactory;
+ String namePrefix = "";
+ if (userFunction instanceof DataGeneratorSource) {
+ RowType outputType = (RowType) LogicalTypeConverter.toVLType(
+ ((InternalTypeInfo) transformation.getOutputType()).toLogicalType());
+ operatorFactory = SimpleOperatorFactory.of(
+ new GlutenStreamSource(
+ new GlutenSourceFunction(
+ new TableScanNode(
+ String.valueOf(transformationId),
+ outputType,
+ new FuzzerTableHandle("connector-fuzzer", 12367),
+ List.of()),
+ outputType,
+ String.valueOf(transformationId))));
+ namePrefix = "Gluten ";
+ } else {
+ operatorFactory = transformation.getOperatorFactory();
+ }
+ streamGraph.addLegacySource(
+ transformationId,
+ slotSharingGroup,
+ transformation.getCoLocationGroupKey(),
+ operatorFactory,
+ null,
+ transformation.getOutputType(),
+ namePrefix + "Source: " + transformation.getName());
+
+ if (transformation.getOperatorFactory() instanceof InputFormatOperatorFactory) {
+ streamGraph.setInputFormat(
+ transformationId,
+ ((InputFormatOperatorFactory<OUT>) transformation.getOperatorFactory())
+ .getInputFormat());
+ }
+
+ final int parallelism =
+ transformation.getParallelism() != ExecutionConfig.PARALLELISM_DEFAULT
+ ? transformation.getParallelism()
+ : executionConfig.getParallelism();
+ streamGraph.setParallelism(
+ transformationId, parallelism, transformation.isParallelismConfigured());
+ streamGraph.setMaxParallelism(transformationId, transformation.getMaxParallelism());
+
+ streamGraph.setSupportsConcurrentExecutionAttempts(
+ transformationId, transformation.isSupportsConcurrentExecutionAttempts());
+
+ return Collections.singleton(transformationId);
+ }
+}
diff --git a/gluten-flink/runtime/src/main/java/org/apache/flink/streaming/runtime/translators/SinkTransformationTranslator.java b/gluten-flink/runtime/src/main/java/org/apache/flink/streaming/runtime/translators/SinkTransformationTranslator.java
new file mode 100644
index 0000000000..51cb262b90
--- /dev/null
+++ b/gluten-flink/runtime/src/main/java/org/apache/flink/streaming/runtime/translators/SinkTransformationTranslator.java
@@ -0,0 +1,556 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.runtime.translators;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.SupportsConcurrentExecutionAttempts;
+import org.apache.flink.api.common.operators.SlotSharingGroup;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.sink2.Sink;
+import org.apache.flink.api.connector.sink2.SupportsCommitter;
+import org.apache.flink.api.dag.Transformation;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.streaming.api.connector.sink2.CommittableMessage;
+import org.apache.flink.streaming.api.connector.sink2.CommittableMessageTypeInfo;
+import org.apache.flink.streaming.api.connector.sink2.StandardSinkTopologies;
+import org.apache.flink.streaming.api.connector.sink2.SupportsPostCommitTopology;
+import org.apache.flink.streaming.api.connector.sink2.SupportsPreCommitTopology;
+import org.apache.flink.streaming.api.connector.sink2.SupportsPreWriteTopology;
+import org.apache.flink.streaming.api.datastream.CustomSinkOperatorUidHashes;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink;
+import org.apache.flink.streaming.api.graph.TransformationTranslator;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
+import org.apache.flink.streaming.api.transformations.OneInputTransformation;
+import org.apache.flink.streaming.api.transformations.PartitionTransformation;
+import org.apache.flink.streaming.api.transformations.PhysicalTransformation;
+import org.apache.flink.streaming.api.transformations.SinkTransformation;
+import org.apache.flink.streaming.api.transformations.StreamExchangeMode;
+import org.apache.flink.streaming.runtime.operators.sink.CommitterOperatorFactory;
+import org.apache.flink.streaming.runtime.operators.sink.SinkWriterOperatorFactory;
+import org.apache.flink.streaming.runtime.partitioner.ForwardPartitioner;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.util.Preconditions;
+
+import io.github.zhztheplayer.velox4j.connector.CommitStrategy;
+import io.github.zhztheplayer.velox4j.connector.DiscardDataTableHandle;
+import io.github.zhztheplayer.velox4j.plan.TableWriteNode;
+import io.github.zhztheplayer.velox4j.serde.Serde;
+import io.github.zhztheplayer.velox4j.type.BigIntType;
+import io.github.zhztheplayer.velox4j.type.RowType;
+import org.apache.gluten.streaming.api.operators.GlutenOneInputOperatorFactory;
+import org.apache.gluten.table.runtime.operators.GlutenCalOperator;
+import org.apache.gluten.util.LogicalTypeConverter;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayDeque;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Optional;
+import java.util.Queue;
+import java.util.Set;
+import java.util.function.BiConsumer;
+import java.util.function.Function;
+
+import static org.apache.flink.util.Preconditions.checkState;
+
+/**
+ * A {@link org.apache.flink.streaming.api.graph.TransformationTranslator} for the {@link
+ * org.apache.flink.streaming.api.transformations.SinkTransformation}.
+ */
+@Internal
+public class SinkTransformationTranslator<Input, Output>
+ implements TransformationTranslator<Output, SinkTransformation<Input, Output>> {
+
+ private static final String COMMITTER_NAME = "Committer";
+ private static final String WRITER_NAME = "Writer";
+
+ @Override
+ public Collection<Integer> translateForBatch(
+ SinkTransformation<Input, Output> transformation, Context context) {
+ return translateInternal(transformation, context, true);
+ }
+
+ @Override
+ public Collection<Integer> translateForStreaming(
+ SinkTransformation<Input, Output> transformation, Context context) {
+ return translateInternal(transformation, context, false);
+ }
+
+ private Collection<Integer> translateInternal(
+ SinkTransformation<Input, Output> transformation, Context context, boolean batch) {
+ SinkExpander<Input> expander =
+ new SinkExpander<>(
+ transformation.getInputStream(),
+ transformation.getSink(),
+ transformation,
+ context,
+ batch);
+ expander.expand();
+ return Collections.emptyList();
+ }
+
+ /**
+ * Expands the FLIP-143 Sink to a subtopology. Each part of the topology is created after the
+ * previous part of the topology has been completely configured by the user. For example, if a
+ * user explicitly sets the parallelism of the sink, each part of the subtopology can rely on
+ * the input having that parallelism.
+ */
+ private static class SinkExpander<T> {
+ private final SinkTransformation<T, ?> transformation;
+ private final Sink<T> sink;
+ private final Context context;
+ private final DataStream<T> inputStream;
+ private final StreamExecutionEnvironment executionEnvironment;
+ private final Optional<Integer> environmentParallelism;
+ private final boolean isBatchMode;
+ private final boolean isCheckpointingEnabled;
+
+ public SinkExpander(
+ DataStream<T> inputStream,
+ Sink<T> sink,
+ SinkTransformation<T, ?> transformation,
+ Context context,
+ boolean isBatchMode) {
+ this.inputStream = inputStream;
+ this.executionEnvironment = inputStream.getExecutionEnvironment();
+ this.environmentParallelism =
+ executionEnvironment
+ .getConfig()
+ .toConfiguration()
+ .getOptional(CoreOptions.DEFAULT_PARALLELISM);
+ this.isCheckpointingEnabled =
+ executionEnvironment.getCheckpointConfig().isCheckpointingEnabled();
+ this.transformation = transformation;
+ this.sink = sink;
+ this.context = context;
+ this.isBatchMode = isBatchMode;
+ }
+
+ private void expand() {
+
+ final int sizeBefore = executionEnvironment.getTransformations().size();
+
+ DataStream<T> prewritten = inputStream;
+
+ if (sink instanceof SupportsPreWriteTopology) {
+ prewritten =
+ adjustTransformations(
+ prewritten,
+ ((SupportsPreWriteTopology<T>) sink)::addPreWriteTopology,
+ true,
+ sink instanceof SupportsConcurrentExecutionAttempts);
+ }
+
+ if (sink instanceof SupportsPreCommitTopology) {
+ Preconditions.checkArgument(
+ sink instanceof SupportsCommitter,
+ "Sink with SupportsPreCommitTopology should implement SupportsCommitter");
+ }
+ if (sink instanceof SupportsPostCommitTopology) {
+ Preconditions.checkArgument(
+ sink instanceof SupportsCommitter,
+ "Sink with SupportsPostCommitTopology should implement SupportsCommitter");
+ }
+
+ if (sink instanceof SupportsCommitter) {
+ addCommittingTopology(sink, prewritten);
+ } else {
+ if (sink instanceof DiscardingSink) {
+ RowType outputType = (RowType) LogicalTypeConverter.toVLType(
+ ((InternalTypeInfo) transformation.getOutputType()).toLogicalType());
+ RowType outType = new RowType(List.of("num"), List.of(new BigIntType()));
+ String plan = Serde.toJson(new TableWriteNode(
+ String.valueOf(transformation.getId()),
+ outputType,
+ outputType.getNames(),
+ null,
+ "connector-fuzzer",
+ new DiscardDataTableHandle(),
+ false,
+ outType,
+ CommitStrategy.NO_COMMIT,
+ null));
+ adjustTransformations(
+ prewritten,
+ input ->
+ input.transform(
+ WRITER_NAME,
+ CommittableMessageTypeInfo.noOutput(),
+ new GlutenOneInputOperatorFactory(
+ new GlutenCalOperator(
+ plan,
+ String.valueOf(transformation.getId()),
+ "",
+ Serde.toJson(outType)
+ ))),
+ false,
+ sink instanceof SupportsConcurrentExecutionAttempts);
+ } else {
+ adjustTransformations(
+ prewritten,
+ input ->
+ input.transform(
+ WRITER_NAME,
+ CommittableMessageTypeInfo.noOutput(),
+ new SinkWriterOperatorFactory<>(sink)),
+ false,
+ sink instanceof SupportsConcurrentExecutionAttempts);
+ }
+ }
+
+ getSinkTransformations(sizeBefore).forEach(context::transform);
+
+ disallowUnalignedCheckpoint(getSinkTransformations(sizeBefore));
+
+ // Remove all added sink subtransformations to avoid duplications and allow additional
+ // expansions
+ while (executionEnvironment.getTransformations().size() > sizeBefore) {
+ executionEnvironment
+ .getTransformations()
+ .remove(executionEnvironment.getTransformations().size() - 1);
+ }
+ }
+
+ private List<Transformation<?>> getSinkTransformations(int sizeBefore) {
+ return executionEnvironment
+ .getTransformations()
+ .subList(sizeBefore, executionEnvironment.getTransformations().size());
+ }
+
+ /**
+ * Disables UC for all connections of operators within the sink expansion. This is necessary
+ * because committables need to be at the respective operators on notifyCheckpointComplete
+ * or else we can't commit all side-effects, which violates the contract of
+ * notifyCheckpointComplete.
+ */
+ private void disallowUnalignedCheckpoint(List<Transformation<?>> sinkTransformations) {
+ Optional<Transformation<?>> writerOpt =
+ sinkTransformations.stream().filter(SinkExpander::isWriter).findFirst();
+ Preconditions.checkState(writerOpt.isPresent(), "Writer transformation not found.");
+ Transformation<?> writer = writerOpt.get();
+ int indexOfWriter = sinkTransformations.indexOf(writer);
+
+ // check all transformation after the writer and recursively disable UC for all inputs
+ // up to the writer
+ Set<Integer> seen = new HashSet<>(writer.getId());
+ Queue<Transformation<?>> pending =
+ new ArrayDeque<>(
+ sinkTransformations.subList(
+ indexOfWriter + 1, sinkTransformations.size()));
+
+ while (!pending.isEmpty()) {
+ Transformation<?> current = pending.poll();
+ seen.add(current.getId());
+
+ for (Transformation<?> input : current.getInputs()) {
+ if (input instanceof PartitionTransformation) {
+ ((PartitionTransformation<?>) input)
+ .getPartitioner()
+ .disableUnalignedCheckpoints();
+ }
+ if (seen.add(input.getId())) {
+ pending.add(input);
+ }
+ }
+ }
+ }
+
+ private static boolean isWriter(Transformation<?> t) {
+ if (!(t instanceof OneInputTransformation)) {
+ return false;
+ }
+ StreamOperatorFactory operatorFactory =
+ ((OneInputTransformation<?, ?>) t).getOperatorFactory();
+ return operatorFactory instanceof SinkWriterOperatorFactory ||
+ operatorFactory instanceof GlutenOneInputOperatorFactory;
+ }
+
+ private <CommT, WriteResultT> void addCommittingTopology(
+ Sink<T> sink, DataStream<T> inputStream) {
+ SupportsCommitter<CommT> committingSink = (SupportsCommitter<CommT>) sink;
+ TypeInformation<CommittableMessage<CommT>> committableTypeInformation =
+ CommittableMessageTypeInfo.of(committingSink::getCommittableSerializer);
+
+ DataStream<CommittableMessage<CommT>> precommitted;
+ if (sink instanceof SupportsPreCommitTopology) {
+ SupportsPreCommitTopology<WriteResultT, CommT> preCommittingSink =
+ (SupportsPreCommitTopology<WriteResultT, CommT>) sink;
+ TypeInformation<CommittableMessage<WriteResultT>> writeResultTypeInformation =
+ CommittableMessageTypeInfo.of(preCommittingSink::getWriteResultSerializer);
+
+ DataStream<CommittableMessage<WriteResultT>> writerResult =
+ addWriter(sink, inputStream, writeResultTypeInformation);
+
+ precommitted =
+ adjustTransformations(
+ writerResult, preCommittingSink::addPreCommitTopology, true, false);
+ } else {
+ precommitted = addWriter(sink, inputStream, committableTypeInformation);
+ }
+
+ DataStream<CommittableMessage<CommT>> committed =
+ adjustTransformations(
+ precommitted,
+ pc ->
+ pc.transform(
+ COMMITTER_NAME,
+ committableTypeInformation,
+ new CommitterOperatorFactory<>(
+ committingSink,
+ isBatchMode,
+ isCheckpointingEnabled)),
+ false,
+ false);
+
+ if (sink instanceof SupportsPostCommitTopology) {
+ DataStream<CommittableMessage<CommT>> postcommitted = addFailOverRegion(committed);
+ adjustTransformations(
+ postcommitted,
+ pc -> {
+ ((SupportsPostCommitTopology<CommT>) sink).addPostCommitTopology(pc);
+ return null;
+ },
+ true,
+ false);
+ }
+ }
+
+ private <WriteResultT> DataStream<CommittableMessage<WriteResultT>> addWriter(
+ Sink<T> sink,
+ DataStream<T> inputStream,
+ TypeInformation<CommittableMessage<WriteResultT>> typeInformation) {
+ DataStream<CommittableMessage<WriteResultT>> written =
+ adjustTransformations(
+ inputStream,
+ input ->
+ input.transform(
+ WRITER_NAME,
+ typeInformation,
+ new SinkWriterOperatorFactory<>(sink)),
+ false,
+ sink instanceof SupportsConcurrentExecutionAttempts);
+
+ return addFailOverRegion(written);
+ }
+
+ /**
+ * Adds a batch exchange that materializes the output first. This is a no-op in STREAMING.
+ */
+ private <I> DataStream<I> addFailOverRegion(DataStream<I> input) {
+ return new DataStream<>(
+ executionEnvironment,
+ new PartitionTransformation<>(
+ input.getTransformation(),
+ new ForwardPartitioner<>(),
+ StreamExchangeMode.BATCH));
+ }
+
+ /**
+ * Since user may set specific parallelism on sub topologies, we have to pay attention to
+ * the priority of parallelism at different levels, i.e. sub topologies customized
+ * parallelism > sinkTransformation customized parallelism > environment customized
+ * parallelism. In order to satisfy this rule and keep these customized parallelism values,
+ * the environment parallelism will be set to be {@link ExecutionConfig#PARALLELISM_DEFAULT}
+ * before adjusting transformations. SubTransformations, constructed after that, will have
+ * either the default value or customized value. In this way, any customized value will be
+ * discriminated from the default value and, for any subTransformation with the default
+ * parallelism value, we will then be able to let it inherit the parallelism value from the
+ * previous sinkTransformation. After the adjustment of transformations is closed, the
+ * environment parallelism will be restored back to its original value to keep the
+ * customized parallelism value at environment level.
+ */
+ private <I, R> R adjustTransformations(
+ DataStream<I> inputStream,
+ Function<DataStream<I>, R> action,
+ boolean isExpandedTopology,
+ boolean supportsConcurrentExecutionAttempts) {
+
+ // Reset the environment parallelism temporarily before adjusting transformations,
+ // we can therefore be aware of any customized parallelism of the sub topology
+ // set by users during the adjustment.
+ executionEnvironment.setParallelism(ExecutionConfig.PARALLELISM_DEFAULT);
+
+ int numTransformsBefore = executionEnvironment.getTransformations().size();
+ R result = action.apply(inputStream);
+ List<Transformation<?>> transformations = executionEnvironment.getTransformations();
+ List<Transformation<?>> expandedTransformations =
+ transformations.subList(numTransformsBefore, transformations.size());
+
+ final CustomSinkOperatorUidHashes operatorsUidHashes =
+ transformation.getSinkOperatorsUidHashes();
+ for (Transformation<?> subTransformation : expandedTransformations) {
+
+ String subUid = subTransformation.getUid();
+ if (isExpandedTopology && subUid != null && !subUid.isEmpty()) {
+ checkState(
+ transformation.getUid() != null && !transformation.getUid().isEmpty(),
+ "Sink "
+ + transformation.getName()
+ + " requires to set a uid since its customized topology"
+ + " has set uid for some operators.");
+ }
+
+ // Set the operator uid hashes to support stateful upgrades without prior uids
+ setOperatorUidHashIfPossible(
+ subTransformation, WRITER_NAME, operatorsUidHashes.getWriterUidHash());
+ setOperatorUidHashIfPossible(
+ subTransformation,
+ COMMITTER_NAME,
+ operatorsUidHashes.getCommitterUidHash());
+ setOperatorUidHashIfPossible(
+ subTransformation,
+ StandardSinkTopologies.GLOBAL_COMMITTER_TRANSFORMATION_NAME,
+ operatorsUidHashes.getGlobalCommitterUidHash());
+
+ concatUid(
+ subTransformation,
+ Transformation::getUid,
+ Transformation::setUid,
+ subTransformation.getName());
+
+ concatProperty(
+ subTransformation,
+ Transformation::getCoLocationGroupKey,
+ Transformation::setCoLocationGroupKey);
+
+ concatProperty(subTransformation, Transformation::getName, Transformation::setName);
+
+ concatProperty(
+ subTransformation,
+ Transformation::getDescription,
+ Transformation::setDescription);
+
+ // handle coLocationGroupKey.
+ String coLocationGroupKey = transformation.getCoLocationGroupKey();
+ if (coLocationGroupKey != null
+ && subTransformation.getCoLocationGroupKey() == null) {
+ subTransformation.setCoLocationGroupKey(coLocationGroupKey);
+ }
+
+ Optional<SlotSharingGroup> ssg = transformation.getSlotSharingGroup();
+
+ if (ssg.isPresent() && !subTransformation.getSlotSharingGroup().isPresent()) {
+ subTransformation.setSlotSharingGroup(ssg.get());
+ }
+
+ // remember that the environment parallelism has been set to be default
+ // at the beginning. SubTransformations, whose parallelism has been
+ // customized, will skip this part. The customized parallelism value set by user
+ // will therefore be kept.
+ if (subTransformation.getParallelism() == ExecutionConfig.PARALLELISM_DEFAULT) {
+ // In this case, the subTransformation does not contain any customized
+ // parallelism value and will therefore inherit the parallelism value
+ // from the sinkTransformation.
+ subTransformation.setParallelism(
+ transformation.getParallelism(),
+ transformation.isParallelismConfigured());
+ }
+
+ if (subTransformation.getMaxParallelism() < 0
+ && transformation.getMaxParallelism() > 0) {
+ subTransformation.setMaxParallelism(transformation.getMaxParallelism());
+ }
+
+ if (subTransformation instanceof PhysicalTransformation) {
+ PhysicalTransformation<?> physicalSubTransformation =
+ (PhysicalTransformation<?>) subTransformation;
+
+ if (transformation.getChainingStrategy() != null) {
+ physicalSubTransformation.setChainingStrategy(
+ transformation.getChainingStrategy());
+ }
+
+ // overrides the supportsConcurrentExecutionAttempts of transformation because
+ // it's not allowed to specify fine-grained concurrent execution attempts yet
+ physicalSubTransformation.setSupportsConcurrentExecutionAttempts(
+ supportsConcurrentExecutionAttempts);
+ }
+ }
+
+ // Restore the previous parallelism of the environment before adjusting transformations
+ if (environmentParallelism.isPresent()) {
+ executionEnvironment.getConfig().setParallelism(environmentParallelism.get());
+ } else {
+ executionEnvironment.getConfig().resetParallelism();
+ }
+
+ return result;
+ }
+
+ private void setOperatorUidHashIfPossible(
+ Transformation<?> transformation,
+ String writerName,
+ @Nullable String operatorUidHash) {
+ if (operatorUidHash == null || !transformation.getName().equals(writerName)) {
+ return;
+ }
+ transformation.setUidHash(operatorUidHash);
+ }
+
+ private void concatUid(
+ Transformation<?> subTransformation,
+ Function<Transformation<?>, String> getter,
+ BiConsumer<Transformation<?>, String> setter,
+ @Nullable String transformationName) {
+ if (transformationName != null && getter.apply(transformation) != null) {
+ // Use the same uid pattern than for Sink V1. We deliberately decided to use the uid
+ // pattern of Flink 1.13 because 1.14 did not have a dedicated committer operator.
+ if (transformationName.equals(COMMITTER_NAME)) {
+ final String committerFormat = "Sink Committer: %s";
+ setter.accept(
+ subTransformation,
+ String.format(committerFormat, getter.apply(transformation)));
+ return;
+ }
+ // Set the writer operator uid to the sinks uid to support state migrations
+ if (transformationName.equals(WRITER_NAME)) {
+ setter.accept(subTransformation, getter.apply(transformation));
+ return;
+ }
+
+ // Use the same uid pattern than for Sink V1 in Flink 1.14.
+ if (transformationName.equals(
+ StandardSinkTopologies.GLOBAL_COMMITTER_TRANSFORMATION_NAME)) {
+ final String committerFormat = "Sink %s Global Committer";
+ setter.accept(
+ subTransformation,
+ String.format(committerFormat, getter.apply(transformation)));
+ return;
+ }
+ }
+ concatProperty(subTransformation, getter, setter);
+ }
+
+ private void concatProperty(
+ Transformation<?> subTransformation,
+ Function<Transformation<?>, String> getter,
+ BiConsumer<Transformation<?>, String> setter) {
+ if (getter.apply(transformation) != null && getter.apply(subTransformation) != null) {
+ setter.accept(
+ subTransformation,
+ getter.apply(transformation) + ": " + getter.apply(subTransformation));
+ }
+ }
+ }
+}
diff --git a/gluten-flink/runtime/src/main/java/org/apache/gluten/streaming/api/operators/GlutenOneInputOperatorFactory.java b/gluten-flink/runtime/src/main/java/org/apache/gluten/streaming/api/operators/GlutenOneInputOperatorFactory.java
new file mode 100644
index 0000000000..280848edcc
--- /dev/null
+++ b/gluten-flink/runtime/src/main/java/org/apache/gluten/streaming/api/operators/GlutenOneInputOperatorFactory.java
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gluten.streaming.api.operators;
+
+import io.github.zhztheplayer.velox4j.plan.PlanNode;
+import io.github.zhztheplayer.velox4j.type.RowType;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
+
+/** One input operator factory for gluten. */
+public class GlutenOneInputOperatorFactory<IN, OUT> extends AbstractStreamOperatorFactory<OUT>
+ implements OneInputStreamOperatorFactory<IN, OUT> {
+ private final GlutenOperator operator;
+
+ public GlutenOneInputOperatorFactory(GlutenOperator operator) {
+ this.operator = operator;
+ }
+
+ public GlutenOperator getOperator() {
+ return operator;
+ }
+
+ @Override
+ public <T extends StreamOperator<OUT>> T createStreamOperator(StreamOperatorParameters<OUT> streamOperatorParameters) {
+ throw new RuntimeException("Not Implemented");
+ }
+
+ @Override
+ public Class<? extends StreamOperator> getStreamOperatorClass(ClassLoader classLoader) {
+ throw new RuntimeException("Not Implemented");
+ }
+}
diff --git a/gluten-flink/runtime/src/main/java/org/apache/gluten/streaming/api/operators/GlutenOperator.java b/gluten-flink/runtime/src/main/java/org/apache/gluten/streaming/api/operators/GlutenOperator.java
new file mode 100644
index 0000000000..4d7238a4eb
--- /dev/null
+++ b/gluten-flink/runtime/src/main/java/org/apache/gluten/streaming/api/operators/GlutenOperator.java
@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gluten.streaming.api.operators;
+
+import io.github.zhztheplayer.velox4j.plan.PlanNode;
+import io.github.zhztheplayer.velox4j.type.RowType;
+
+/** Interface for all gluten operators. */
+public interface GlutenOperator {
+ public PlanNode getPlanNode();
+ public RowType getOutputType();
+ public String getId();
+}
diff --git a/gluten-flink/runtime/src/main/java/org/apache/gluten/streaming/api/operators/GlutenStreamSource.java b/gluten-flink/runtime/src/main/java/org/apache/gluten/streaming/api/operators/GlutenStreamSource.java
new file mode 100644
index 0000000000..ac551338ff
--- /dev/null
+++ b/gluten-flink/runtime/src/main/java/org/apache/gluten/streaming/api/operators/GlutenStreamSource.java
@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gluten.streaming.api.operators;
+
+import org.apache.gluten.table.runtime.operators.GlutenSourceFunction;
+
+import io.github.zhztheplayer.velox4j.plan.PlanNode;
+import io.github.zhztheplayer.velox4j.type.RowType;
+import org.apache.flink.streaming.api.operators.StreamSource;
+
+/** Legacy stream source operator in gluten, which will call Velox to run. */
+public class GlutenStreamSource extends StreamSource
+ implements GlutenOperator {
+ private final GlutenSourceFunction sourceFunction;
+
+ public GlutenStreamSource(GlutenSourceFunction function) {
+ super(function);
+ sourceFunction = function;
+ }
+
+ @Override
+ public void open() throws Exception {
+ super.open();
+ }
+
+ @Override
+ public PlanNode getPlanNode() {
+ return sourceFunction.getPlanNode();
+ }
+
+ @Override
+ public RowType getOutputType() {
+ return sourceFunction.getOutputType();
+ }
+
+ @Override
+ public String getId() {
+ return sourceFunction.getId();
+ }
+}
diff --git a/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenCalOperator.java b/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenCalOperator.java
new file mode 100644
index 0000000000..6940d002c1
--- /dev/null
+++ b/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenCalOperator.java
@@ -0,0 +1,134 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gluten.table.runtime.operators;
+
+import io.github.zhztheplayer.velox4j.connector.ExternalStream;
+import io.github.zhztheplayer.velox4j.connector.ExternalStreamConnectorSplit;
+import io.github.zhztheplayer.velox4j.iterator.CloseableIterator;
+import io.github.zhztheplayer.velox4j.iterator.DownIterators;
+import io.github.zhztheplayer.velox4j.iterator.UpIterators;
+import io.github.zhztheplayer.velox4j.query.BoundSplit;
+import io.github.zhztheplayer.velox4j.serde.Serde;
+import io.github.zhztheplayer.velox4j.type.RowType;
+import org.apache.gluten.streaming.api.operators.GlutenOperator;
+import org.apache.gluten.vectorized.FlinkRowToVLVectorConvertor;
+
+import io.github.zhztheplayer.velox4j.Velox4j;
+import io.github.zhztheplayer.velox4j.config.Config;
+import io.github.zhztheplayer.velox4j.config.ConnectorConfig;
+import io.github.zhztheplayer.velox4j.data.RowVector;
+import io.github.zhztheplayer.velox4j.memory.AllocationListener;
+import io.github.zhztheplayer.velox4j.memory.MemoryManager;
+import io.github.zhztheplayer.velox4j.plan.PlanNode;
+import io.github.zhztheplayer.velox4j.query.Query;
+import io.github.zhztheplayer.velox4j.session.Session;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.operators.TableStreamOperator;
+
+import java.util.List;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+/** Calculate operator in gluten, which will call Velox to run. */
+public class GlutenCalOperator extends TableStreamOperator<RowData>
+ implements OneInputStreamOperator<RowData, RowData>, GlutenOperator {
+
+ private final String glutenPlan;
+ private final String id;
+ private final String inputType;
+ private final String outputType;
+
+ private StreamRecord outElement = null;
+
+ private Session session;
+ private Query query;
+ private BlockingQueue<RowVector> inputQueue;
+ BufferAllocator allocator;
+
+ public GlutenCalOperator(String plan, String id, String inputType, String outputType) {
+ this.glutenPlan = plan;
+ this.id = id;
+ this.inputType = inputType;
+ this.outputType = outputType;
+ }
+
+ @Override
+ public void open() throws Exception {
+ super.open();
+ outElement = new StreamRecord(null);
+ session = Velox4j.newSession(MemoryManager.create(AllocationListener.NOOP));
+
+ inputQueue = new LinkedBlockingQueue<>();
+ ExternalStream es =
+ session.externalStreamOps().bind(DownIterators.fromBlockingQueue(inputQueue));
+ List<BoundSplit> splits = List.of(
+ new BoundSplit(
+ id,
+ -1,
+ new ExternalStreamConnectorSplit("connector-external-stream", es.id())));
+ PlanNode filter = Serde.fromJson(glutenPlan, PlanNode.class);
+ query = new Query(filter, splits, Config.empty(), ConnectorConfig.empty());
+ allocator = new RootAllocator(Long.MAX_VALUE);
+
+ }
+
+ @Override
+ public void processElement(StreamRecord<RowData> element) {
+ inputQueue.add(
+ FlinkRowToVLVectorConvertor.fromRowData(
+ element.getValue(),
+ allocator,
+ session,
+ Serde.fromJson(inputType, RowType.class)));
+ CloseableIterator<RowVector> result =
+ UpIterators.asJavaIterator(session.queryOps().execute(query));
+ if (result.hasNext()) {
+ List<RowData> rows = FlinkRowToVLVectorConvertor.toRowData(
+ result.next(),
+ allocator,
+ session,
+ Serde.fromJson(outputType, RowType.class));
+ for (RowData row : rows) {
+ output.collect(outElement.replace(row));
+ }
+ }
+ }
+
+ @Override
+ public PlanNode getPlanNode() {
+ // TODO: support wartermark operator
+ if (glutenPlan == "Watermark") {
+ return null;
+ }
+ return Serde.fromJson(glutenPlan, PlanNode.class);
+ }
+
+ @Override
+ public RowType getOutputType() {
+ return Serde.fromJson(outputType, RowType.class);
+ }
+
+ @Override
+ public String getId() {
+ return id;
+ }
+}
diff --git a/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenSourceFunction.java b/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenSourceFunction.java
new file mode 100644
index 0000000000..687fba577b
--- /dev/null
+++ b/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenSourceFunction.java
@@ -0,0 +1,99 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gluten.table.runtime.operators;
+
+import io.github.zhztheplayer.velox4j.connector.FuzzerConnectorSplit;
+import io.github.zhztheplayer.velox4j.query.BoundSplit;
+import org.apache.gluten.vectorized.FlinkRowToVLVectorConvertor;
+
+import io.github.zhztheplayer.velox4j.Velox4j;
+import io.github.zhztheplayer.velox4j.config.Config;
+import io.github.zhztheplayer.velox4j.config.ConnectorConfig;
+import io.github.zhztheplayer.velox4j.data.RowVector;
+import io.github.zhztheplayer.velox4j.iterator.CloseableIterator;
+import io.github.zhztheplayer.velox4j.iterator.UpIterators;
+import io.github.zhztheplayer.velox4j.memory.AllocationListener;
+import io.github.zhztheplayer.velox4j.memory.MemoryManager;
+import io.github.zhztheplayer.velox4j.plan.PlanNode;
+import io.github.zhztheplayer.velox4j.query.Query;
+import io.github.zhztheplayer.velox4j.session.Session;
+import io.github.zhztheplayer.velox4j.type.RowType;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
+import org.apache.flink.table.data.RowData;
+
+import java.util.List;
+
+/** Gluten legacy source function, call velox plan to execute. */
+public class GlutenSourceFunction extends RichParallelSourceFunction<RowData> {
+
+ private final PlanNode planNode;
+ private final RowType outputType;
+ private final String id;
+ private volatile boolean isRunning = true;
+
+ private Session session;
+ private Query query;
+ BufferAllocator allocator;
+
+ public GlutenSourceFunction(PlanNode planNode, RowType outputType, String id) {
+ this.planNode = planNode;
+ this.outputType = outputType;
+ this.id = id;
+ }
+
+ public PlanNode getPlanNode() {
+ return planNode;
+ }
+
+ public RowType getOutputType() { return outputType; }
+
+ public String getId() { return id; }
+
+ @Override
+ public void run(SourceContext<RowData> sourceContext) throws Exception {
+ final List<BoundSplit> splits = List.of(new BoundSplit(
+ id,
+ -1,
+ new FuzzerConnectorSplit("connector-fuzzer", 1000)));
+ session = Velox4j.newSession(MemoryManager.create(AllocationListener.NOOP));
+ query = new Query(planNode, splits, Config.empty(), ConnectorConfig.empty());
+ allocator = new RootAllocator(Long.MAX_VALUE);
+
+ while (isRunning) {
+ CloseableIterator<RowVector> result =
+ UpIterators.asJavaIterator(session.queryOps().execute(query));
+ if (result.hasNext()) {
+ List<RowData> rows = FlinkRowToVLVectorConvertor.toRowData(
+ result.next(),
+ allocator,
+ session,
+ outputType);
+ for (RowData row : rows) {
+ sourceContext.collect(row);
+ }
+ }
+ }
+ }
+
+ @Override
+ public void cancel() {
+ isRunning = false;
+ }
+}
diff --git a/gluten-flink/runtime/src/main/java/org/apache/gluten/util/LogicalTypeConverter.java b/gluten-flink/runtime/src/main/java/org/apache/gluten/util/LogicalTypeConverter.java
new file mode 100644
index 0000000000..c9fa4e0526
--- /dev/null
+++ b/gluten-flink/runtime/src/main/java/org/apache/gluten/util/LogicalTypeConverter.java
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.gluten.util;
+
+import io.github.zhztheplayer.velox4j.type.IntegerType;
+import io.github.zhztheplayer.velox4j.type.Type;
+import org.apache.flink.table.types.logical.BigIntType;
+import org.apache.flink.table.types.logical.DecimalType;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.TimestampType;
+import org.apache.flink.table.types.logical.VarCharType;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+/** Convertor to convert Flink LogicalType to velox data Type */
+public class LogicalTypeConverter {
+
+ public static Type toVLType(LogicalType logicalType) {
+ if (logicalType instanceof RowType) {
+ RowType flinkRowType = (RowType) logicalType;
+ List<Type> fieldTypes = flinkRowType.getChildren().stream().
+ map(LogicalTypeConverter::toVLType).
+ collect(Collectors.toList());
+ return new io.github.zhztheplayer.velox4j.type.RowType(
+ flinkRowType.getFieldNames(),
+ fieldTypes);
+ } else if (logicalType instanceof IntType) {
+ return new IntegerType();
+ } else if (logicalType instanceof BigIntType) {
+ return new io.github.zhztheplayer.velox4j.type.BigIntType();
+ } else if (logicalType instanceof VarCharType) {
+ return new io.github.zhztheplayer.velox4j.type.VarCharType();
+ } else if (logicalType instanceof TimestampType) {
+ // TODO: may need precision
+ return new io.github.zhztheplayer.velox4j.type.TimestampType();
+ } else if (logicalType instanceof DecimalType) {
+ DecimalType decimalType = (DecimalType) logicalType;
+ return new io.github.zhztheplayer.velox4j.type.DecimalType(
+ decimalType.getPrecision(),
+ decimalType.getScale());
+ } else {
+ throw new RuntimeException("Unsupported logical type: " + logicalType);
+ }
+ }
+}
diff --git a/gluten-flink/runtime/src/main/java/org/apache/gluten/vectorized/FlinkRowToVLVectorConvertor.java b/gluten-flink/runtime/src/main/java/org/apache/gluten/vectorized/FlinkRowToVLVectorConvertor.java
new file mode 100644
index 0000000000..894ce903cc
--- /dev/null
+++ b/gluten-flink/runtime/src/main/java/org/apache/gluten/vectorized/FlinkRowToVLVectorConvertor.java
@@ -0,0 +1,105 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.gluten.vectorized;
+
+import io.github.zhztheplayer.velox4j.data.RowVector;
+import io.github.zhztheplayer.velox4j.session.Session;
+import io.github.zhztheplayer.velox4j.type.BigIntType;
+import io.github.zhztheplayer.velox4j.type.IntegerType;
+import io.github.zhztheplayer.velox4j.type.RowType;
+import io.github.zhztheplayer.velox4j.type.Type;
+import io.github.zhztheplayer.velox4j.type.VarCharType;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.vector.BigIntVector;
+import org.apache.arrow.vector.IntVector;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.VarCharVector;
+import org.apache.arrow.vector.table.Table;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.binary.BinaryStringData;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/** Converter between velox RowVector and Flink RowData. */
+public class FlinkRowToVLVectorConvertor {
+
+ public static RowVector fromRowData(
+ RowData row,
+ BufferAllocator allocator,
+ Session session,
+ RowType rowType) {
+ // TODO: support more types
+ List<FieldVector> arrowVectors = new ArrayList<>(rowType.size());
+ for (int i = 0; i < rowType.size(); i++) {
+ Type fieldType = rowType.getChildren().get(i);
+ if (fieldType instanceof IntegerType) {
+ IntVector intVector = new IntVector(rowType.getNames().get(i), allocator);
+ intVector.setSafe(0, row.getInt(i));
+ intVector.setValueCount(1);
+ arrowVectors.add(i, intVector);
+ } else if (fieldType instanceof BigIntType) {
+ BigIntVector bigIntVector = new BigIntVector(rowType.getNames().get(i), allocator);
+ bigIntVector.setSafe(0, row.getLong(i));
+ bigIntVector.setValueCount(1);
+ arrowVectors.add(i, bigIntVector);
+ } else if (fieldType instanceof VarCharType) {
+ VarCharVector stringVector = new VarCharVector(rowType.getNames().get(i), allocator);
+ stringVector.setSafe(0, row.getString(i).toBytes());
+ stringVector.setValueCount(1);
+ arrowVectors.add(i, stringVector);
+ } else {
+ throw new RuntimeException("Unsupported field type: " + fieldType);
+ }
+ }
+ return session.arrowOps().fromArrowTable(allocator, new Table(arrowVectors));
+ }
+
+ public static List<RowData> toRowData(
+ RowVector rowVector,
+ BufferAllocator allocator,
+ Session session,
+ RowType rowType) {
+ // TODO: support more types
+ FieldVector fieldVector = session.arrowOps().toArrowVector(
+ allocator,
+ rowVector.loadedVector());
+ List<RowData> rowDatas = new ArrayList<>(rowVector.getSize());
+ for (int j = 0; j < rowVector.getSize(); j++) {
+ List<Object> fieldValues = new ArrayList<>(rowType.size());
+ for (int i = 0; i < rowType.size(); i++) {
+ Type fieldType = rowType.getChildren().get(i);
+ if (fieldType instanceof IntegerType) {
+ fieldValues.add(i, ((IntVector) fieldVector.getChildrenFromFields().get(i)).get(j));
+ } else if (fieldType instanceof BigIntType) {
+ fieldValues.add(i, ((BigIntVector) fieldVector.getChildrenFromFields().get(i)).get(j));
+ } else if (fieldType instanceof VarCharType) {
+ fieldValues.add(
+ i,
+ BinaryStringData.fromBytes(
+ ((VarCharVector) fieldVector.getChildrenFromFields().get(i)).get(j)));
+ } else {
+ throw new RuntimeException("Unsupported field type: " + fieldType);
+ }
+ }
+ rowDatas.add(GenericRowData.of(fieldValues.toArray()));
+ }
+ return rowDatas;
+ }
+
+}
diff --git a/gluten-flink/runtime/src/main/java/org/apache/gluten/vectorized/VLVectorIterator.java b/gluten-flink/runtime/src/main/java/org/apache/gluten/vectorized/VLVectorIterator.java
new file mode 100644
index 0000000000..363bdfb423
--- /dev/null
+++ b/gluten-flink/runtime/src/main/java/org/apache/gluten/vectorized/VLVectorIterator.java
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.gluten.vectorized;
+
+import io.github.zhztheplayer.velox4j.data.RowVector;
+
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+
+/** Iterator for velox RowVector. */
+public class VLVectorIterator implements Iterator<RowVector> {
+
+ private final List<RowVector> rows;
+
+ public VLVectorIterator() {
+ this.rows = new LinkedList<>();
+ }
+
+ public boolean hasNext() {
+ return !rows.isEmpty();
+ }
+
+ public RowVector next() {
+ if (!hasNext()) {
+ return null;
+ }
+ return rows.remove(0);
+ }
+
+ public void addRow(RowVector row) {
+ rows.add(row);
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org