[Improve][Flink] Remove useless stageType. (#5650)

apache · Oct 20, 2023 · 4341c7c · 4341c7c
1 parent d98f182
commit 4341c7c
Show file tree

Hide file tree

Showing 8 changed files with 4 additions and 87 deletions.
diff --git a/release-note.md b/release-note.md
@@ -97,6 +97,7 @@
 - [Core] [Starter] Optimize code structure & remove redundant code (#4525)
 - [Core] [Translation] [Flink] Optimize code structure & remove redundant code (#4527)
 - [Core] [Starter] Add check of sink and source config to avoid null pointer exception. (#4734)
+- [Core] [Flink] Remove useless stage type related codes. (#5650)
 
 ### Connector-V2
 

diff --git a/.../main/java/org/apache/seatunnel/core/starter/flink/execution/FlinkRuntimeEnvironment.java b/.../main/java/org/apache/seatunnel/core/starter/flink/execution/FlinkRuntimeEnvironment.java
@@ -20,7 +20,6 @@
 import org.apache.seatunnel.shade.com.typesafe.config.Config;
 
 import org.apache.seatunnel.api.env.EnvCommonOptions;
-import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
 import org.apache.seatunnel.common.Constants;
 import org.apache.seatunnel.common.config.CheckResult;
 import org.apache.seatunnel.common.constants.JobMode;
@@ -52,11 +51,8 @@
 
 import java.net.URL;
 import java.util.ArrayList;
-import java.util.LinkedHashMap;
 import java.util.List;
-import java.util.Map;
 import java.util.Objects;
-import java.util.Optional;
 import java.util.stream.Collectors;
 
 @Slf4j
@@ -68,8 +64,6 @@ public class FlinkRuntimeEnvironment implements RuntimeEnvironment {
     private StreamExecutionEnvironment environment;
 
     private StreamTableEnvironment tableEnvironment;
-    private Map<String, SeaTunnelRowType> stagedTypes = new LinkedHashMap<>();
-    private Optional<SeaTunnelRowType> defaultType = Optional.empty();
     private JobMode jobMode;
 
     private String jobName = Constants.LOGO;
@@ -339,24 +333,6 @@ public void registerResultTable(
                 name, tableEnvironment.fromChangelogStream(dataStream));
     }
 
-    public void stageType(String tblName, SeaTunnelRowType type) {
-        stagedTypes.put(tblName, type);
-    }
-
-    public void stageDefaultType(SeaTunnelRowType type) {
-        this.defaultType = Optional.of(type);
-    }
-
-    public Optional<SeaTunnelRowType> type(String tblName) {
-        return stagedTypes.containsKey(tblName)
-                ? Optional.of(stagedTypes.get(tblName))
-                : Optional.empty();
-    }
-
-    public Optional<SeaTunnelRowType> defaultType() {
-        return this.defaultType;
-    }
-
     public static FlinkRuntimeEnvironment getInstance(Config config) {
         if (INSTANCE == null) {
             synchronized (FlinkRuntimeEnvironment.class) {

diff --git a/...src/main/java/org/apache/seatunnel/core/starter/flink/execution/SinkExecuteProcessor.java b/...src/main/java/org/apache/seatunnel/core/starter/flink/execution/SinkExecuteProcessor.java
@@ -101,7 +101,7 @@ public List<DataStreamTableInfo> execute(List<DataStreamTableInfo> upstreamDataS
                                         sinkConfig.getString(PLUGIN_NAME.key())),
                                 sinkConfig);
                 sink.setJobContext(jobContext);
-                SeaTunnelRowType sourceType = initSourceType(sinkConfig, stream.getDataStream());
+                SeaTunnelRowType sourceType = stream.getCatalogTable().getSeaTunnelRowType();
                 sink.setTypeInfo(sourceType);
             } else {
                 TableSinkFactoryContext context =

diff --git a/...rg/apache/seatunnel/core/starter/flink/execution/FlinkAbstractPluginExecuteProcessor.java b/...rg/apache/seatunnel/core/starter/flink/execution/FlinkAbstractPluginExecuteProcessor.java
@@ -20,12 +20,10 @@
 import org.apache.seatunnel.shade.com.typesafe.config.Config;
 
 import org.apache.seatunnel.api.common.JobContext;
-import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
 import org.apache.seatunnel.common.utils.ReflectionUtils;
 import org.apache.seatunnel.common.utils.SeaTunnelException;
 import org.apache.seatunnel.core.starter.execution.PluginExecuteProcessor;
 import org.apache.seatunnel.core.starter.flink.utils.TableUtil;
-import org.apache.seatunnel.translation.flink.utils.TypeConverterUtils;
 
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.table.api.Table;
@@ -135,36 +133,6 @@ protected void registerAppendStream(Config pluginConfig) {
         }
     }
 
-    protected void stageType(Config pluginConfig, SeaTunnelRowType type) {
-        if (!flinkRuntimeEnvironment.defaultType().isPresent()) {
-            flinkRuntimeEnvironment.stageDefaultType(type);
-        }
-
-        if (pluginConfig.hasPath("result_table_name")) {
-            String tblName = pluginConfig.getString("result_table_name");
-            flinkRuntimeEnvironment.stageType(tblName, type);
-        }
-    }
-
-    protected Optional<SeaTunnelRowType> sourceType(Config pluginConfig) {
-        if (pluginConfig.hasPath(SOURCE_TABLE_NAME)) {
-            String tblName = pluginConfig.getString(SOURCE_TABLE_NAME);
-            return flinkRuntimeEnvironment.type(tblName);
-        } else {
-            return flinkRuntimeEnvironment.defaultType();
-        }
-    }
-
-    protected SeaTunnelRowType initSourceType(Config sinkConfig, DataStream<Row> stream) {
-        SeaTunnelRowType sourceType =
-                sourceType(sinkConfig)
-                        .orElseGet(
-                                () ->
-                                        (SeaTunnelRowType)
-                                                TypeConverterUtils.convert(stream.getType()));
-        return sourceType;
-    }
-
     protected abstract List<T> initializePlugins(
             List<URL> jarPaths, List<? extends Config> pluginConfigs);
 }
diff --git a/.../main/java/org/apache/seatunnel/core/starter/flink/execution/FlinkRuntimeEnvironment.java b/.../main/java/org/apache/seatunnel/core/starter/flink/execution/FlinkRuntimeEnvironment.java
@@ -20,7 +20,6 @@
 import org.apache.seatunnel.shade.com.typesafe.config.Config;
 
 import org.apache.seatunnel.api.env.EnvCommonOptions;
-import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
 import org.apache.seatunnel.common.Constants;
 import org.apache.seatunnel.common.config.CheckResult;
 import org.apache.seatunnel.common.constants.JobMode;
@@ -52,11 +51,8 @@
 
 import java.net.URL;
 import java.util.ArrayList;
-import java.util.LinkedHashMap;
 import java.util.List;
-import java.util.Map;
 import java.util.Objects;
-import java.util.Optional;
 import java.util.stream.Collectors;
 
 @Slf4j
@@ -69,9 +65,6 @@ public class FlinkRuntimeEnvironment implements RuntimeEnvironment {
 
     private StreamTableEnvironment tableEnvironment;
 
-    private Map<String, SeaTunnelRowType> stagedTypes = new LinkedHashMap<>();
-    private Optional<SeaTunnelRowType> defaultType = Optional.empty();
-
     private JobMode jobMode;
 
     private String jobName = Constants.LOGO;
@@ -341,24 +334,6 @@ public void registerResultTable(
                 name, tableEnvironment.fromChangelogStream(dataStream));
     }
 
-    public void stageType(String tblName, SeaTunnelRowType type) {
-        stagedTypes.put(tblName, type);
-    }
-
-    public void stageDefaultType(SeaTunnelRowType type) {
-        this.defaultType = Optional.of(type);
-    }
-
-    public Optional<SeaTunnelRowType> type(String tblName) {
-        return stagedTypes.containsKey(tblName)
-                ? Optional.of(stagedTypes.get(tblName))
-                : Optional.empty();
-    }
-
-    public Optional<SeaTunnelRowType> defaultType() {
-        return this.defaultType;
-    }
-
     public static FlinkRuntimeEnvironment getInstance(Config config) {
         if (INSTANCE == null) {
             synchronized (FlinkRuntimeEnvironment.class) {

diff --git a/...src/main/java/org/apache/seatunnel/core/starter/flink/execution/SinkExecuteProcessor.java b/...src/main/java/org/apache/seatunnel/core/starter/flink/execution/SinkExecuteProcessor.java
@@ -102,7 +102,7 @@ public List<DataStreamTableInfo> execute(List<DataStreamTableInfo> upstreamDataS
                                         sinkConfig.getString(PLUGIN_NAME.key())),
                                 sinkConfig);
                 sink.setJobContext(jobContext);
-                SeaTunnelRowType sourceType = initSourceType(sinkConfig, stream.getDataStream());
+                SeaTunnelRowType sourceType = stream.getCatalogTable().getSeaTunnelRowType();
                 sink.setTypeInfo(sourceType);
             } else {
                 TableSinkFactoryContext context =

diff --git a/...c/main/java/org/apache/seatunnel/core/starter/flink/execution/SourceExecuteProcessor.java b/...c/main/java/org/apache/seatunnel/core/starter/flink/execution/SourceExecuteProcessor.java
@@ -24,7 +24,6 @@
 import org.apache.seatunnel.api.source.SeaTunnelSource;
 import org.apache.seatunnel.api.source.SupportCoordinate;
 import org.apache.seatunnel.api.table.factory.TableSourceFactory;
-import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
 import org.apache.seatunnel.common.Constants;
 import org.apache.seatunnel.core.starter.enums.PluginType;
 import org.apache.seatunnel.core.starter.execution.PluginUtil;
@@ -95,7 +94,6 @@ public List<DataStreamTableInfo> execute(List<DataStreamTableInfo> upstreamDataS
                             sourceFunction,
                             "SeaTunnel " + internalSource.getClass().getSimpleName(),
                             bounded);
-            stageType(pluginConfig, (SeaTunnelRowType) internalSource.getProducedType());
 
             if (pluginConfig.hasPath(CommonOptions.PARALLELISM.key())) {
                 int parallelism = pluginConfig.getInt(CommonOptions.PARALLELISM.key());

diff --git a/...ain/java/org/apache/seatunnel/core/starter/flink/execution/TransformExecuteProcessor.java b/...ain/java/org/apache/seatunnel/core/starter/flink/execution/TransformExecuteProcessor.java
@@ -91,11 +91,10 @@ public List<DataStreamTableInfo> execute(List<DataStreamTableInfo> upstreamDataS
                 ConfigValidator.of(context.getOptions()).validate(factory.optionRule());
                 SeaTunnelTransform transform = factory.createTransform(context).createTransform();
 
-                SeaTunnelRowType sourceType = initSourceType(pluginConfig, stream.getDataStream());
+                SeaTunnelRowType sourceType = stream.getCatalogTable().getSeaTunnelRowType();
                 transform.setJobContext(jobContext);
                 DataStream<Row> inputStream =
                         flinkTransform(sourceType, transform, stream.getDataStream());
-                stageType(pluginConfig, transform.getProducedCatalogTable().getSeaTunnelRowType());
                 registerResultTable(pluginConfig, inputStream);
                 upstreamDataStreams.add(
                         new DataStreamTableInfo(