callbackFunc,
+ ParameterTool parameter) {
+ this.context = context;
+ this.parameter = parameter;
+ this.statementSet = context.createStatementSet();
+ this.callback =
+ r -> {
+ if (callbackFunc != null) {
+ callbackFunc.accept(r);
+ } else {
+ LOG.info(r);
+ }
+ };
+ }
+ }
+}
diff --git a/streampark-flink/streampark-flink-shims/streampark-flink-shims-base-v2/src/main/java/org/apache/streampark/flink/core/FlinkStreamTableTraitV2.java b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base-v2/src/main/java/org/apache/streampark/flink/core/FlinkStreamTableTraitV2.java
new file mode 100644
index 0000000000..63b5c2d39e
--- /dev/null
+++ b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base-v2/src/main/java/org/apache/streampark/flink/core/FlinkStreamTableTraitV2.java
@@ -0,0 +1,691 @@
+/*
+ * 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.streampark.flink.core;
+
+import org.apache.streampark.common.util.Utils;
+
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.api.common.RuntimeExecutionMode;
+import org.apache.flink.api.common.cache.DistributedCache;
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.common.io.FileInputFormat;
+import org.apache.flink.api.common.io.InputFormat;
+import org.apache.flink.api.connector.source.Source;
+import org.apache.flink.api.connector.source.SourceSplit;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.core.execution.JobListener;
+import org.apache.flink.streaming.api.CheckpointingMode;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.CheckpointConfig;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.source.FileProcessingMode;
+import org.apache.flink.streaming.api.graph.StreamGraph;
+import org.apache.flink.table.api.CompiledPlan;
+import org.apache.flink.table.api.ExplainDetail;
+import org.apache.flink.table.api.ExplainFormat;
+import org.apache.flink.table.api.PlanReference;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.api.TableDescriptor;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.CatalogDescriptor;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.functions.ScalarFunction;
+import org.apache.flink.table.functions.UserDefinedFunction;
+import org.apache.flink.table.module.Module;
+import org.apache.flink.table.module.ModuleEntry;
+import org.apache.flink.table.resource.ResourceUri;
+import org.apache.flink.table.types.AbstractDataType;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.ParameterTool;
+import org.apache.flink.util.SplittableIterator;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Optional;
+import java.util.function.Consumer;
+
+/**
+ * Integration API of stream and table environments for Flink 2.x.
+ *
+ * Once a Table has been converted to a DataStream, the DataStream job must be executed using the
+ * execute method of the StreamExecutionEnvironment.
+ */
+public abstract class FlinkStreamTableTraitV2 implements StreamTableEnvironment {
+
+ public final ParameterTool parameter;
+
+ private final StreamExecutionEnvironment streamEnv;
+
+ private final StreamTableEnvironment tableEnv;
+
+ /** Whether a table has been converted to a DataStream. */
+ public boolean isConvertedToDataStream;
+
+ protected FlinkStreamTableTraitV2(
+ ParameterTool parameter,
+ StreamExecutionEnvironment streamEnv,
+ StreamTableEnvironment tableEnv) {
+ this.parameter = parameter;
+ this.streamEnv = streamEnv;
+ this.tableEnv = tableEnv;
+ }
+
+ protected StreamExecutionEnvironment getStreamEnv() {
+ return streamEnv;
+ }
+
+ protected StreamTableEnvironment getStreamTableEnv() {
+ return tableEnv;
+ }
+
+ /** Recommended API to start tasks. */
+ public JobExecutionResult start() {
+ return start(null);
+ }
+
+ public JobExecutionResult start(String name) {
+ String appName = FlinkParameterUtils.getAppName(parameter, name, true);
+ return execute(appName);
+ }
+
+ /** @deprecated Retained for backward compatibility with legacy Flink Table API. */
+ @Deprecated(since = "2.1.0", forRemoval = false)
+ public JobExecutionResult execute(String jobName) {
+ Utils.printLogo("FlinkStreamTable " + jobName + " Starting...");
+ if (isConvertedToDataStream) {
+ try {
+ return streamEnv.execute(jobName);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ return null;
+ }
+
+ public void sql(String sql) {
+ sql(sql, null);
+ }
+
+ public void sql(String sql, Consumer callback) {
+ FlinkSqlExecutor.executeSql(sql, parameter, this, callback);
+ }
+
+ public StreamExecutionEnvironment getJavaEnv() {
+ return streamEnv;
+ }
+
+ public List> $getCachedFiles() {
+ return streamEnv.getCachedFiles();
+ }
+
+ public List $getJobListeners() {
+ return streamEnv.getJobListeners();
+ }
+
+ public void $setParallelism(int parallelism) {
+ streamEnv.setParallelism(parallelism);
+ }
+
+ public StreamExecutionEnvironment $setRuntimeMode(RuntimeExecutionMode deployMode) {
+ return streamEnv.setRuntimeMode(deployMode);
+ }
+
+ public void $setMaxParallelism(int maxParallelism) {
+ streamEnv.setMaxParallelism(maxParallelism);
+ }
+
+ public int $getParallelism() {
+ return streamEnv.getParallelism();
+ }
+
+ public int $getMaxParallelism() {
+ return streamEnv.getMaxParallelism();
+ }
+
+ public StreamExecutionEnvironment $setBufferTimeout(long timeoutMillis) {
+ return streamEnv.setBufferTimeout(timeoutMillis);
+ }
+
+ public long $getBufferTimeout() {
+ return streamEnv.getBufferTimeout();
+ }
+
+ public StreamExecutionEnvironment $disableOperatorChaining() {
+ return streamEnv.disableOperatorChaining();
+ }
+
+ public CheckpointConfig $getCheckpointConfig() {
+ return streamEnv.getCheckpointConfig();
+ }
+
+ public StreamExecutionEnvironment $enableCheckpointing(long interval, CheckpointingMode mode) {
+ return streamEnv.enableCheckpointing(interval, mode);
+ }
+
+ public StreamExecutionEnvironment $enableCheckpointing(long interval) {
+ return streamEnv.enableCheckpointing(interval);
+ }
+
+ public CheckpointingMode $getCheckpointingMode() {
+ return streamEnv.getCheckpointingMode();
+ }
+
+ public void $configure(ReadableConfig configuration, ClassLoader classLoader) {
+ streamEnv.configure(configuration, classLoader);
+ }
+
+ public DataStream $fromData(T data) {
+ return streamEnv.fromData(data);
+ }
+
+ public DataStream $fromSequence(long from, long to) {
+ return streamEnv.fromSequence(from, to);
+ }
+
+ public DataStream $fromCollection(Collection data) {
+ return streamEnv.fromCollection(data);
+ }
+
+ public DataStream $fromParallelCollection(SplittableIterator data, Class clazz) {
+ return streamEnv.fromParallelCollection(data, clazz);
+ }
+
+ public DataStream $readFile(FileInputFormat inputFormat, String filePath) {
+ return streamEnv.readFile(inputFormat, filePath);
+ }
+
+ public DataStream $readFile(
+ FileInputFormat inputFormat,
+ String filePath,
+ FileProcessingMode watchType,
+ long interval) {
+ return streamEnv.readFile(inputFormat, filePath, watchType, interval);
+ }
+
+ public DataStream $socketTextStream(
+ String hostname, int port, char delimiter, long maxRetry) {
+ return streamEnv.socketTextStream(hostname, port, delimiter, maxRetry);
+ }
+
+ public DataStream $createInput(InputFormat inputFormat) {
+ return streamEnv.createInput(inputFormat);
+ }
+
+ public DataStream $fromSource(
+ Source source,
+ WatermarkStrategy watermarkStrategy,
+ String sourceName) {
+ return streamEnv.fromSource(source, watermarkStrategy, sourceName);
+ }
+
+ public void $registerJobListener(JobListener jobListener) {
+ streamEnv.registerJobListener(jobListener);
+ }
+
+ public void $clearJobListeners() {
+ streamEnv.clearJobListeners();
+ }
+
+ public JobClient $executeAsync() {
+ try {
+ return streamEnv.executeAsync();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ public JobClient $executeAsync(String jobName) {
+ try {
+ return streamEnv.executeAsync(jobName);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ public String $getExecutionPlan() {
+ return streamEnv.getExecutionPlan();
+ }
+
+ public StreamGraph $getStreamGraph() {
+ return streamEnv.getStreamGraph();
+ }
+
+ public StreamExecutionEnvironment $getWrappedStreamExecutionEnvironment() {
+ return streamEnv;
+ }
+
+ public void $registerCachedFile(String filePath, String name) {
+ streamEnv.registerCachedFile(filePath, name);
+ }
+
+ public void $registerCachedFile(String filePath, String name, boolean executable) {
+ streamEnv.registerCachedFile(filePath, name, executable);
+ }
+
+ public boolean $isUnalignedCheckpointsEnabled() {
+ return streamEnv.isUnalignedCheckpointsEnabled();
+ }
+
+ public boolean $isForceUnalignedCheckpoints() {
+ return streamEnv.isForceUnalignedCheckpoints();
+ }
+
+ @Override
+ public Table fromDataStream(DataStream dataStream, Schema schema) {
+ return tableEnv.fromDataStream(dataStream, schema);
+ }
+
+ @Override
+ public Table fromChangelogStream(DataStream dataStream) {
+ return tableEnv.fromChangelogStream(dataStream);
+ }
+
+ @Override
+ public Table fromChangelogStream(DataStream dataStream, Schema schema) {
+ return tableEnv.fromChangelogStream(dataStream, schema);
+ }
+
+ @Override
+ public Table fromChangelogStream(
+ DataStream dataStream, Schema schema, ChangelogMode changelogMode) {
+ return tableEnv.fromChangelogStream(dataStream, schema, changelogMode);
+ }
+
+ @Override
+ public void createTemporaryView(String path, DataStream dataStream, Schema schema) {
+ tableEnv.createTemporaryView(path, dataStream, schema);
+ }
+
+ @Override
+ public DataStream toDataStream(Table table) {
+ isConvertedToDataStream = true;
+ return tableEnv.toDataStream(table);
+ }
+
+ @Override
+ public DataStream toDataStream(Table table, Class targetClass) {
+ isConvertedToDataStream = true;
+ return tableEnv.toDataStream(table, targetClass);
+ }
+
+ @Override
+ public DataStream toDataStream(Table table, AbstractDataType targetDataType) {
+ isConvertedToDataStream = true;
+ return tableEnv.toDataStream(table, targetDataType);
+ }
+
+ @Override
+ public DataStream toChangelogStream(Table table) {
+ isConvertedToDataStream = true;
+ return tableEnv.toChangelogStream(table);
+ }
+
+ @Override
+ public DataStream toChangelogStream(Table table, Schema targetSchema) {
+ isConvertedToDataStream = true;
+ return tableEnv.toChangelogStream(table, targetSchema);
+ }
+
+ @Override
+ public DataStream toChangelogStream(
+ Table table, Schema targetSchema, ChangelogMode changelogMode) {
+ isConvertedToDataStream = true;
+ return tableEnv.toChangelogStream(table, targetSchema, changelogMode);
+ }
+
+ @Override
+ public DataStream toAppendStream(Table table, Class clazz) {
+ isConvertedToDataStream = true;
+ return tableEnv.toAppendStream(table, clazz);
+ }
+
+ @Override
+ public DataStream> toRetractStream(Table table, Class clazz) {
+ isConvertedToDataStream = true;
+ return tableEnv.toRetractStream(table, clazz);
+ }
+
+ @Override
+ public void createCatalog(String catalogName, CatalogDescriptor catalogDescriptor) {
+ tableEnv.createCatalog(catalogName, catalogDescriptor);
+ }
+
+ @Override
+ public void useModules(String... moduleNames) {
+ tableEnv.useModules(moduleNames);
+ }
+
+ @Override
+ public void createFunction(
+ String path, String className, List resourceUris) {
+ tableEnv.createFunction(path, className, resourceUris);
+ }
+
+ @Override
+ public void createFunction(
+ String path,
+ String className,
+ List resourceUris,
+ boolean ignoreIfExists) {
+ tableEnv.createFunction(path, className, resourceUris, ignoreIfExists);
+ }
+
+ @Override
+ public void createTemporaryFunction(
+ String path, String className, List resourceUris) {
+ tableEnv.createTemporaryFunction(path, className, resourceUris);
+ }
+
+ @Override
+ public void createTemporarySystemFunction(
+ String name, String className, List resourceUris) {
+ tableEnv.createTemporarySystemFunction(name, className, resourceUris);
+ }
+
+ @Override
+ public void createTemporaryTable(String path, TableDescriptor descriptor) {
+ tableEnv.createTemporaryTable(path, descriptor);
+ }
+
+ @Override
+ public void createTable(String path, TableDescriptor descriptor) {
+ tableEnv.createTable(path, descriptor);
+ }
+
+ @Override
+ public Table from(TableDescriptor descriptor) {
+ return tableEnv.from(descriptor);
+ }
+
+ @Override
+ public ModuleEntry[] listFullModules() {
+ return tableEnv.listFullModules();
+ }
+
+ @Override
+ public String[] listTables(String catalogName, String databaseName) {
+ return tableEnv.listTables(catalogName, databaseName);
+ }
+
+ @Override
+ public String explainSql(
+ String statement, ExplainFormat format, ExplainDetail... extraDetails) {
+ return tableEnv.explainSql(statement, format, extraDetails);
+ }
+
+ @Override
+ public CompiledPlan loadPlan(PlanReference planReference) throws TableException {
+ return tableEnv.loadPlan(planReference);
+ }
+
+ @Override
+ public CompiledPlan compilePlanSql(String statement) throws TableException {
+ return tableEnv.compilePlanSql(statement);
+ }
+
+ @Override
+ public Table fromDataStream(DataStream dataStream) {
+ return tableEnv.fromDataStream(dataStream);
+ }
+
+ @Override
+ public Table fromDataStream(DataStream dataStream, Expression... fields) {
+ return tableEnv.fromDataStream(dataStream, fields);
+ }
+
+ @Override
+ public void createTemporaryView(String path, DataStream dataStream) {
+ tableEnv.createTemporaryView(path, dataStream);
+ }
+
+ @Override
+ public void createTemporaryView(
+ String path, DataStream dataStream, Expression... fields) {
+ tableEnv.createTemporaryView(path, dataStream, fields);
+ }
+
+ @Override
+ public Table fromValues(Expression... values) {
+ return tableEnv.fromValues(values);
+ }
+
+ @Override
+ public Table fromValues(AbstractDataType rowType, Expression... values) {
+ return tableEnv.fromValues(rowType, values);
+ }
+
+ @Override
+ public Table fromValues(Iterable values) {
+ return tableEnv.fromValues(values);
+ }
+
+ @Override
+ public Table fromValues(AbstractDataType rowType, Iterable values) {
+ return tableEnv.fromValues(rowType, values);
+ }
+
+ @Override
+ public void registerCatalog(String catalogName, Catalog catalog) {
+ tableEnv.registerCatalog(catalogName, catalog);
+ }
+
+ @Override
+ public Optional getCatalog(String catalogName) {
+ return tableEnv.getCatalog(catalogName);
+ }
+
+ @Override
+ public void loadModule(String moduleName, Module module) {
+ tableEnv.loadModule(moduleName, module);
+ }
+
+ @Override
+ public void unloadModule(String moduleName) {
+ tableEnv.unloadModule(moduleName);
+ }
+
+ @Override
+ public void createTemporarySystemFunction(
+ String name, Class functionClass) {
+ tableEnv.createTemporarySystemFunction(name, functionClass);
+ }
+
+ @Override
+ public void createTemporarySystemFunction(
+ String name, UserDefinedFunction functionInstance) {
+ tableEnv.createTemporarySystemFunction(name, functionInstance);
+ }
+
+ @Override
+ public boolean dropTemporarySystemFunction(String name) {
+ return tableEnv.dropTemporarySystemFunction(name);
+ }
+
+ @Override
+ public void createFunction(String path, Class functionClass) {
+ tableEnv.createFunction(path, functionClass);
+ }
+
+ @Override
+ public void createFunction(
+ String path,
+ Class functionClass,
+ boolean ignoreIfExists) {
+ tableEnv.createFunction(path, functionClass, ignoreIfExists);
+ }
+
+ @Override
+ public boolean dropFunction(String path) {
+ return tableEnv.dropFunction(path);
+ }
+
+ @Override
+ public void createTemporaryFunction(
+ String path, Class functionClass) {
+ tableEnv.createTemporaryFunction(path, functionClass);
+ }
+
+ @Override
+ public void createTemporaryFunction(String path, UserDefinedFunction functionInstance) {
+ tableEnv.createTemporaryFunction(path, functionInstance);
+ }
+
+ @Override
+ public boolean dropTemporaryFunction(String path) {
+ return tableEnv.dropTemporaryFunction(path);
+ }
+
+ @Override
+ public void createTemporaryView(String path, Table view) {
+ tableEnv.createTemporaryView(path, view);
+ }
+
+ @Override
+ public Table from(String path) {
+ return tableEnv.from(path);
+ }
+
+ @Override
+ public String[] listCatalogs() {
+ return tableEnv.listCatalogs();
+ }
+
+ @Override
+ public String[] listModules() {
+ return tableEnv.listModules();
+ }
+
+ @Override
+ public String[] listDatabases() {
+ return tableEnv.listDatabases();
+ }
+
+ @Override
+ public String[] listTables() {
+ return tableEnv.listTables();
+ }
+
+ @Override
+ public String[] listViews() {
+ return tableEnv.listViews();
+ }
+
+ @Override
+ public String[] listTemporaryTables() {
+ return tableEnv.listTemporaryTables();
+ }
+
+ @Override
+ public String[] listTemporaryViews() {
+ return tableEnv.listTemporaryViews();
+ }
+
+ @Override
+ public String[] listUserDefinedFunctions() {
+ return tableEnv.listUserDefinedFunctions();
+ }
+
+ @Override
+ public String[] listFunctions() {
+ return tableEnv.listFunctions();
+ }
+
+ @Override
+ public boolean dropTemporaryTable(String path) {
+ return tableEnv.dropTemporaryTable(path);
+ }
+
+ @Override
+ public boolean dropTemporaryView(String path) {
+ return tableEnv.dropTemporaryView(path);
+ }
+
+ @Override
+ public String explainSql(String statement, ExplainDetail... extraDetails) {
+ return tableEnv.explainSql(statement, extraDetails);
+ }
+
+ @Override
+ public Table sqlQuery(String query) {
+ return tableEnv.sqlQuery(query);
+ }
+
+ @Override
+ public TableResult executeSql(String statement) {
+ return tableEnv.executeSql(statement);
+ }
+
+ @Override
+ public String getCurrentCatalog() {
+ return tableEnv.getCurrentCatalog();
+ }
+
+ @Override
+ public void useCatalog(String catalogName) {
+ tableEnv.useCatalog(catalogName);
+ }
+
+ @Override
+ public String getCurrentDatabase() {
+ return tableEnv.getCurrentDatabase();
+ }
+
+ @Override
+ public void useDatabase(String databaseName) {
+ tableEnv.useDatabase(databaseName);
+ }
+
+ @Override
+ public TableConfig getConfig() {
+ return tableEnv.getConfig();
+ }
+
+ @Override
+ public String[] getCompletionHints(String statement, int position) {
+ return tableEnv.getCompletionHints(statement, position);
+ }
+
+ /** @deprecated Retained for backward compatibility with legacy Flink Table API. */
+ @Deprecated(since = "2.1.0", forRemoval = false)
+ @Override
+ public Table scan(String... tablePath) {
+ return tableEnv.scan(tablePath);
+ }
+
+ /** @deprecated Retained for backward compatibility with legacy Flink Table API. */
+ @Deprecated(since = "2.1.0", forRemoval = false)
+ @Override
+ public void registerTable(String name, Table table) {
+ tableEnv.registerTable(name, table);
+ }
+
+ /** @deprecated Retained for backward compatibility with legacy Flink Table API. */
+ @Deprecated(since = "2.1.0", forRemoval = false)
+ @Override
+ public void registerFunction(String name, ScalarFunction function) {
+ tableEnv.registerFunction(name, function);
+ }
+}
diff --git a/streampark-flink/streampark-flink-shims/streampark-flink-shims-base-v2/src/main/java/org/apache/streampark/flink/core/FlinkStreamingInitializerV2.java b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base-v2/src/main/java/org/apache/streampark/flink/core/FlinkStreamingInitializerV2.java
new file mode 100644
index 0000000000..f412ce4af8
--- /dev/null
+++ b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base-v2/src/main/java/org/apache/streampark/flink/core/FlinkStreamingInitializerV2.java
@@ -0,0 +1,193 @@
+/*
+ * 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.streampark.flink.core;
+
+import org.apache.streampark.common.conf.ConfigKeys;
+import org.apache.streampark.common.util.DeflaterUtils;
+import org.apache.streampark.common.util.FileUtils;
+import org.apache.streampark.common.util.HdfsUtils;
+import org.apache.streampark.common.util.PropertiesUtils;
+import org.apache.streampark.flink.core.conf.FlinkConfiguration;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.util.ParameterTool;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+/** Initializes Flink streaming execution environment from application arguments. */
+public class FlinkStreamingInitializerV2 {
+
+ final String[] args;
+
+ StreamEnvConfigFunction javaStreamEnvConfFunc;
+
+ private FlinkConfiguration configuration;
+
+ private StreamExecutionEnvironment streamEnv;
+
+ FlinkStreamingInitializerV2(String[] args) {
+ this.args = args;
+ }
+
+ public static StreamingInitResult initialize(String[] args, StreamEnvConfigFunction config) {
+ FlinkStreamingInitializerV2 flinkInitializer = new FlinkStreamingInitializerV2(args);
+ flinkInitializer.javaStreamEnvConfFunc = config;
+ return new StreamingInitResult(
+ flinkInitializer.getConfiguration().parameter, flinkInitializer.getStreamEnv());
+ }
+
+ public static StreamingInitResult initialize(StreamEnvConfig args) {
+ FlinkStreamingInitializerV2 flinkInitializer = new FlinkStreamingInitializerV2(args.args);
+ flinkInitializer.javaStreamEnvConfFunc = args.conf;
+ return new StreamingInitResult(
+ flinkInitializer.getConfiguration().parameter, flinkInitializer.getStreamEnv());
+ }
+
+ ParameterTool getParameter() {
+ return getConfiguration().parameter;
+ }
+
+ FlinkConfiguration getConfiguration() {
+ if (configuration == null) {
+ configuration = initParameter();
+ }
+ return configuration;
+ }
+
+ StreamExecutionEnvironment getStreamEnv() {
+ if (streamEnv == null) {
+ streamEnv =
+ StreamExecutionEnvironment.getExecutionEnvironment(
+ getConfiguration().envConfig);
+ if (javaStreamEnvConfFunc != null) {
+ javaStreamEnvConfFunc.configuration(streamEnv, getConfiguration().parameter);
+ }
+ streamEnv.getConfig().setGlobalJobParameters(getConfiguration().parameter);
+ }
+ return streamEnv;
+ }
+
+ FlinkConfiguration initParameter() {
+ ParameterTool argsMap = ParameterTool.fromArgs(args);
+ String configFile = argsMap.get(ConfigKeys.KEY_APP_CONF(), null);
+ if (configFile == null || configFile.isEmpty()) {
+ throw new ExceptionInInitializerError(
+ "[StreamPark] Usage:can't find config,please set \"--conf $path \" in main arguments");
+ }
+ Map configMap = parseConfig(configFile);
+ Map properConf =
+ extractConfigByPrefix(configMap, ConfigKeys.KEY_FLINK_PROPERTY_PREFIX());
+ Map appConf =
+ extractConfigByPrefix(configMap, ConfigKeys.KEY_APP_PREFIX());
+
+ ParameterTool parameter =
+ ParameterTool.fromSystemProperties()
+ .mergeWith(ParameterTool.fromMap(properConf))
+ .mergeWith(ParameterTool.fromMap(appConf))
+ .mergeWith(argsMap);
+
+ Configuration envConfig = Configuration.fromMap(properConf);
+ return new FlinkConfiguration(parameter, envConfig, null);
+ }
+
+ Map parseConfig(String config) {
+ Map map;
+ if (config.startsWith("yaml://")) {
+ map = PropertiesUtils.fromYamlText(DeflaterUtils.unzipString(config.substring(7)));
+ } else if (config.startsWith("conf://")) {
+ map = PropertiesUtils.fromHoconText(DeflaterUtils.unzipString(config.substring(7)));
+ } else if (config.startsWith("prop://")) {
+ map =
+ PropertiesUtils.fromPropertiesText(
+ DeflaterUtils.unzipString(config.substring(7)));
+ } else if (config.startsWith("hdfs://")) {
+ try {
+ String text = HdfsUtils.read(config);
+ map = readConfig(config, text);
+ } catch (IOException e) {
+ throw new IllegalArgumentException(
+ "[StreamPark] Failed to read application config from HDFS: " + config, e);
+ }
+ } else {
+ File file = new File(config);
+ if (!file.exists()) {
+ throw new IllegalArgumentException(
+ "[StreamPark] Usage: application config file: "
+ + file
+ + " is not found!!!");
+ }
+ try {
+ map = readConfig(config, FileUtils.readFile(file));
+ } catch (IOException e) {
+ throw new IllegalArgumentException(
+ "[StreamPark] Failed to read application config file: " + config, e);
+ }
+ }
+ Map filtered = new HashMap<>();
+ map.forEach(
+ (key, value) -> {
+ if (value != null && !value.isEmpty()) {
+ filtered.put(key, value);
+ }
+ });
+ return filtered;
+ }
+
+ private Map readConfig(String config, String text) {
+ String format = config.substring(config.lastIndexOf('.') + 1).toLowerCase();
+ switch (format) {
+ case "yml":
+ case "yaml":
+ return PropertiesUtils.fromYamlText(text);
+ case "conf":
+ return PropertiesUtils.fromHoconText(text);
+ case "properties":
+ return PropertiesUtils.fromPropertiesText(text);
+ default:
+ throw new IllegalArgumentException(
+ "[StreamPark] Usage: application config file error,must be [yaml|conf|properties]");
+ }
+ }
+
+ Map extractConfigByPrefix(Map configMap, String prefix) {
+ Map map = new HashMap<>();
+ configMap.forEach(
+ (key, value) -> {
+ if (key.startsWith(prefix)) {
+ map.put(key.substring(prefix.length()), value);
+ }
+ });
+ return map;
+ }
+
+ /** Streaming initialization result. */
+ public static final class StreamingInitResult {
+
+ public final ParameterTool parameter;
+ public final StreamExecutionEnvironment streamEnv;
+
+ StreamingInitResult(ParameterTool parameter, StreamExecutionEnvironment streamEnv) {
+ this.parameter = parameter;
+ this.streamEnv = streamEnv;
+ }
+ }
+}
diff --git a/streampark-flink/streampark-flink-shims/streampark-flink-shims-base-v2/src/main/java/org/apache/streampark/flink/core/FlinkTableInitializerV2.java b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base-v2/src/main/java/org/apache/streampark/flink/core/FlinkTableInitializerV2.java
new file mode 100644
index 0000000000..7799c1e2b1
--- /dev/null
+++ b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base-v2/src/main/java/org/apache/streampark/flink/core/FlinkTableInitializerV2.java
@@ -0,0 +1,297 @@
+/*
+ * 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.streampark.flink.core;
+
+import org.apache.streampark.common.conf.ConfigKeys;
+import org.apache.streampark.common.enums.PlannerType;
+import org.apache.streampark.common.util.DeflaterUtils;
+import org.apache.streampark.common.util.PropertiesUtils;
+import org.apache.streampark.common.util.StreamParkLoggerFactory;
+import org.apache.streampark.flink.core.conf.FlinkConfiguration;
+
+import org.apache.streampark.shaded.org.slf4j.Logger;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.util.ParameterTool;
+
+import java.io.File;
+import java.lang.reflect.Method;
+import java.util.HashMap;
+import java.util.Map;
+
+/** Initializes Flink table and stream-table environments from application arguments. */
+public class FlinkTableInitializerV2 extends FlinkStreamingInitializerV2 {
+
+ private static final Logger LOG =
+ StreamParkLoggerFactory.loggerFactory()
+ .getLogger(FlinkTableInitializerV2.class.getName());
+
+ private TableEnvConfigFunction javaTableEnvConfFunc;
+
+ private EnvironmentSettings.Builder envSettingsBuilder;
+
+ private TableEnvironment tableEnv;
+
+ private StreamTableEnvironment streamTableEnv;
+
+ FlinkTableInitializerV2(String[] args) {
+ super(args);
+ }
+
+ public static TableInitResult initialize(TableEnvConfig args) {
+ FlinkTableInitializerV2 flinkInitializer = new FlinkTableInitializerV2(args.args);
+ flinkInitializer.javaTableEnvConfFunc = args.conf;
+ return new TableInitResult(
+ flinkInitializer.getConfiguration().parameter, flinkInitializer.getTableEnv());
+ }
+
+ public static StreamTableInitResult initialize(StreamTableEnvConfig args) {
+ FlinkTableInitializerV2 flinkInitializer = new FlinkTableInitializerV2(args.args);
+ flinkInitializer.javaStreamEnvConfFunc = args.streamConfig;
+ flinkInitializer.javaTableEnvConfFunc = args.tableConfig;
+ return new StreamTableInitResult(
+ flinkInitializer.getConfiguration().parameter,
+ flinkInitializer.getStreamEnv(),
+ flinkInitializer.getStreamTableEnv());
+ }
+
+ public static StreamTableInitResult initialize(
+ String[] args,
+ StreamEnvConfigFunction streamConfig,
+ TableEnvConfigFunction tableConfig) {
+ FlinkTableInitializerV2 flinkInitializer = new FlinkTableInitializerV2(args);
+ flinkInitializer.javaStreamEnvConfFunc = streamConfig;
+ flinkInitializer.javaTableEnvConfFunc = tableConfig;
+ return new StreamTableInitResult(
+ flinkInitializer.getConfiguration().parameter,
+ flinkInitializer.getStreamEnv(),
+ flinkInitializer.getStreamTableEnv());
+ }
+
+ TableEnvironment getTableEnv() {
+ if (tableEnv == null) {
+ LOG.info("job working in batch mode");
+ EnvironmentSettings.Builder builder = getEnvSettingsBuilder();
+ builder.inBatchMode();
+ tableEnv =
+ FlinkParameterUtils.setAppName(
+ TableEnvironment.create(builder.build()), getParameter());
+ applyTableEnvConfig(tableEnv.getConfig());
+ }
+ return tableEnv;
+ }
+
+ StreamTableEnvironment getStreamTableEnv() {
+ if (streamTableEnv == null) {
+ LOG.info("components should work in streaming mode");
+ EnvironmentSettings.Builder builder = getEnvSettingsBuilder();
+ builder.inStreamingMode();
+ EnvironmentSettings setting = builder.build();
+
+ if (javaStreamEnvConfFunc != null) {
+ javaStreamEnvConfFunc.configuration(getStreamEnv(), getParameter());
+ }
+ streamTableEnv =
+ FlinkParameterUtils.setAppName(
+ StreamTableEnvironment.create(getStreamEnv(), setting), getParameter());
+ applyTableEnvConfig(streamTableEnv.getConfig());
+ }
+ return streamTableEnv;
+ }
+
+ private void applyTableEnvConfig(TableConfig config) {
+ if (javaTableEnvConfFunc != null) {
+ javaTableEnvConfFunc.configuration(config, getParameter());
+ }
+ }
+
+ private EnvironmentSettings.Builder getEnvSettingsBuilder() {
+ if (envSettingsBuilder == null) {
+ envSettingsBuilder = buildEnvSettings(getParameter());
+ }
+ return envSettingsBuilder;
+ }
+
+ private EnvironmentSettings.Builder buildEnvSettings(ParameterTool parameter) {
+ EnvironmentSettings.Builder builder = EnvironmentSettings.newInstance();
+
+ PlannerType plannerType = PlannerType.BLINK;
+ String plannerName = parameter.get(ConfigKeys.KEY_FLINK_TABLE_PLANNER(), null);
+ if (plannerName != null && !plannerName.isEmpty()) {
+ try {
+ plannerType = PlannerType.withName(plannerName);
+ } catch (IllegalArgumentException e) {
+ plannerType = PlannerType.BLINK;
+ }
+ }
+
+ switch (plannerType) {
+ case BLINK:
+ invokePlannerMethod(builder, "useBlinkPlanner", "blinkPlanner will be used.");
+ break;
+ case OLD:
+ invokePlannerMethod(builder, "useOldPlanner", "useOldPlanner will be used.");
+ break;
+ case ANY:
+ invokePlannerMethod(builder, "useAnyPlanner", "useAnyPlanner will be used.");
+ break;
+ default:
+ break;
+ }
+
+ String flinkConf = parameter.get(ConfigKeys.KEY_FLINK_CONF(), null);
+ if (flinkConf == null || flinkConf.isEmpty()) {
+ throw new ExceptionInInitializerError(
+ "[StreamPark] Usage:can't find config,please set \"--flink.conf $conf \" in main arguments");
+ }
+ builder.withConfiguration(
+ Configuration.fromMap(
+ PropertiesUtils.fromYamlText(DeflaterUtils.unzipString(flinkConf))));
+
+ String catalog = parameter.get(ConfigKeys.KEY_FLINK_TABLE_CATALOG(), null);
+ String database = parameter.get(ConfigKeys.KEY_FLINK_TABLE_DATABASE(), null);
+ if (catalog != null && database != null) {
+ LOG.info("with built in catalog: {}", catalog);
+ LOG.info("with built in database: {}", database);
+ builder.withBuiltInCatalogName(catalog);
+ builder.withBuiltInDatabaseName(database);
+ } else if (catalog != null) {
+ LOG.info("with built in catalog: {}", catalog);
+ builder.withBuiltInCatalogName(catalog);
+ } else if (database != null) {
+ LOG.info("with built in database: {}", database);
+ builder.withBuiltInDatabaseName(database);
+ }
+ return builder;
+ }
+
+ private void invokePlannerMethod(
+ EnvironmentSettings.Builder builder, String methodName, String successMessage) {
+ try {
+ Method method = builder.getClass().getDeclaredMethod(methodName);
+ method.setAccessible(true);
+ method.invoke(builder);
+ if (successMessage != null) {
+ LOG.info(successMessage);
+ }
+ } catch (NoSuchMethodException e) {
+ LOG.warn("{} deprecated", methodName);
+ } catch (ReflectiveOperationException e) {
+ LOG.warn("Failed to invoke {} on EnvironmentSettings.Builder", methodName, e);
+ }
+ }
+
+ @Override
+ FlinkConfiguration initParameter() {
+ ParameterTool argsMap = ParameterTool.fromArgs(args);
+ String configFile = argsMap.get(ConfigKeys.KEY_APP_CONF(), null);
+ FlinkConfiguration configuration;
+ if (configFile == null || configFile.isEmpty()) {
+ LOG.warn("Usage:can't find config,you can set \"--conf $path \" in main arguments");
+ ParameterTool parameter = ParameterTool.fromSystemProperties().mergeWith(argsMap);
+ configuration =
+ new FlinkConfiguration(parameter, new Configuration(), new Configuration());
+ } else {
+ Map configMap = parseConfig(configFile);
+ Map sqlConf = new HashMap<>();
+ configMap.forEach(
+ (key, value) -> {
+ if (key.startsWith(ConfigKeys.KEY_SQL_PREFIX())) {
+ sqlConf.put(key.substring(ConfigKeys.KEY_SQL_PREFIX().length()), value);
+ }
+ });
+
+ Map properConf =
+ extractConfigByPrefix(configMap, ConfigKeys.KEY_FLINK_PROPERTY_PREFIX());
+ Map appConf =
+ extractConfigByPrefix(configMap, ConfigKeys.KEY_APP_PREFIX());
+ Map tableConf =
+ extractConfigByPrefix(configMap, ConfigKeys.KEY_FLINK_TABLE_PREFIX());
+
+ Configuration tableConfig = Configuration.fromMap(tableConf);
+ Configuration envConfig = Configuration.fromMap(properConf);
+
+ ParameterTool parameter =
+ ParameterTool.fromSystemProperties()
+ .mergeWith(ParameterTool.fromMap(properConf))
+ .mergeWith(ParameterTool.fromMap(tableConf))
+ .mergeWith(ParameterTool.fromMap(appConf))
+ .mergeWith(ParameterTool.fromMap(sqlConf))
+ .mergeWith(argsMap);
+
+ configuration = new FlinkConfiguration(parameter, envConfig, tableConfig);
+ }
+
+ String flinkSql = configuration.parameter.get(ConfigKeys.KEY_FLINK_SQL(), null);
+ if (flinkSql == null) {
+ return configuration;
+ }
+
+ try {
+ String value = DeflaterUtils.unzipString(flinkSql);
+ return configuration.withParameter(
+ configuration.parameter.mergeWith(
+ ParameterTool.fromMap(
+ Map.of(ConfigKeys.KEY_FLINK_SQL(), value))));
+ } catch (Exception ignored) {
+ File sqlFile = new File(flinkSql);
+ try {
+ Map value =
+ PropertiesUtils.fromYamlFile(sqlFile.getAbsolutePath());
+ return configuration.withParameter(
+ configuration.parameter.mergeWith(ParameterTool.fromMap(value)));
+ } catch (Exception e) {
+ throw new IllegalArgumentException("[StreamPark] init sql error." + e, e);
+ }
+ }
+ }
+
+ /** Table initialization result. */
+ public static final class TableInitResult {
+
+ public final ParameterTool parameter;
+ public final TableEnvironment tableEnv;
+
+ public TableInitResult(ParameterTool parameter, TableEnvironment tableEnv) {
+ this.parameter = parameter;
+ this.tableEnv = tableEnv;
+ }
+ }
+
+ /** Stream-table initialization result. */
+ public static final class StreamTableInitResult {
+
+ public final ParameterTool parameter;
+ public final StreamExecutionEnvironment streamEnv;
+ public final StreamTableEnvironment streamTableEnv;
+
+ public StreamTableInitResult(
+ ParameterTool parameter,
+ StreamExecutionEnvironment streamEnv,
+ StreamTableEnvironment streamTableEnv) {
+ this.parameter = parameter;
+ this.streamEnv = streamEnv;
+ this.streamTableEnv = streamTableEnv;
+ }
+ }
+}
diff --git a/streampark-flink/streampark-flink-shims/streampark-flink-shims-base-v2/src/main/java/org/apache/streampark/flink/core/FlinkTableTrait.java b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base-v2/src/main/java/org/apache/streampark/flink/core/FlinkTableTrait.java
new file mode 100644
index 0000000000..e655b37347
--- /dev/null
+++ b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base-v2/src/main/java/org/apache/streampark/flink/core/FlinkTableTrait.java
@@ -0,0 +1,382 @@
+/*
+ * 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.streampark.flink.core;
+
+import org.apache.streampark.common.util.Utils;
+
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.table.api.CompiledPlan;
+import org.apache.flink.table.api.ExplainDetail;
+import org.apache.flink.table.api.ExplainFormat;
+import org.apache.flink.table.api.PlanReference;
+import org.apache.flink.table.api.StatementSet;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.api.TableDescriptor;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.CatalogDescriptor;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.functions.ScalarFunction;
+import org.apache.flink.table.functions.UserDefinedFunction;
+import org.apache.flink.table.module.Module;
+import org.apache.flink.table.module.ModuleEntry;
+import org.apache.flink.table.resource.ResourceUri;
+import org.apache.flink.table.types.AbstractDataType;
+import org.apache.flink.util.ParameterTool;
+
+import java.util.List;
+import java.util.Optional;
+
+/** Base table environment trait with SQL execution helpers. */
+public abstract class FlinkTableTrait implements TableEnvironment {
+
+ public final ParameterTool parameter;
+
+ private final TableEnvironment tableEnv;
+
+ protected FlinkTableTrait(ParameterTool parameter, TableEnvironment tableEnv) {
+ this.parameter = parameter;
+ this.tableEnv = tableEnv;
+ }
+
+ protected TableEnvironment getTableEnv() {
+ return tableEnv;
+ }
+
+ public JobExecutionResult start() {
+ String appName = FlinkParameterUtils.getAppName(parameter, true);
+ return execute(appName);
+ }
+
+ public JobExecutionResult execute(String jobName) {
+ Utils.printLogo("FlinkTable " + jobName + " Starting...");
+ return null;
+ }
+
+ public void sql(String sql) {
+ FlinkSqlExecutor.executeSql(sql, parameter, this);
+ }
+
+ @Override
+ public Table fromValues(Expression... values) {
+ return tableEnv.fromValues(values);
+ }
+
+ @Override
+ public Table fromValues(AbstractDataType rowType, Expression... values) {
+ return tableEnv.fromValues(rowType, values);
+ }
+
+ @Override
+ public Table fromValues(Iterable values) {
+ return tableEnv.fromValues(values);
+ }
+
+ @Override
+ public Table fromValues(AbstractDataType rowType, Iterable values) {
+ return tableEnv.fromValues(rowType, values);
+ }
+
+ @Override
+ public void createCatalog(String catalogName, CatalogDescriptor catalogDescriptor) {
+ tableEnv.createCatalog(catalogName, catalogDescriptor);
+ }
+
+ @Override
+ public void useModules(String... moduleNames) {
+ tableEnv.useModules(moduleNames);
+ }
+
+ @Override
+ public void createFunction(
+ String path, String className, List resourceUris) {
+ tableEnv.createFunction(path, className, resourceUris);
+ }
+
+ @Override
+ public void createFunction(
+ String path,
+ String className,
+ List resourceUris,
+ boolean ignoreIfExists) {
+ tableEnv.createFunction(path, className, resourceUris, ignoreIfExists);
+ }
+
+ @Override
+ public void createTemporaryFunction(
+ String path, String className, List resourceUris) {
+ tableEnv.createTemporaryFunction(path, className, resourceUris);
+ }
+
+ @Override
+ public void createTemporarySystemFunction(
+ String name, String className, List resourceUris) {
+ tableEnv.createTemporarySystemFunction(name, className, resourceUris);
+ }
+
+ @Override
+ public void createTemporaryTable(String path, TableDescriptor descriptor) {
+ tableEnv.createTemporaryTable(path, descriptor);
+ }
+
+ @Override
+ public void createTable(String path, TableDescriptor descriptor) {
+ tableEnv.createTable(path, descriptor);
+ }
+
+ @Override
+ public Table from(TableDescriptor descriptor) {
+ return tableEnv.from(descriptor);
+ }
+
+ @Override
+ public ModuleEntry[] listFullModules() {
+ return tableEnv.listFullModules();
+ }
+
+ @Override
+ public String[] listTables(String catalogName, String databaseName) {
+ return tableEnv.listTables(catalogName, databaseName);
+ }
+
+ @Override
+ public String explainSql(
+ String statement, ExplainFormat format, ExplainDetail... extraDetails) {
+ return tableEnv.explainSql(statement, format, extraDetails);
+ }
+
+ @Override
+ public CompiledPlan loadPlan(PlanReference planReference) throws TableException {
+ return tableEnv.loadPlan(planReference);
+ }
+
+ @Override
+ public CompiledPlan compilePlanSql(String statement) throws TableException {
+ return tableEnv.compilePlanSql(statement);
+ }
+
+ @Override
+ public void registerCatalog(String catalogName, Catalog catalog) {
+ tableEnv.registerCatalog(catalogName, catalog);
+ }
+
+ @Override
+ public Optional getCatalog(String catalogName) {
+ return tableEnv.getCatalog(catalogName);
+ }
+
+ @Override
+ public void loadModule(String moduleName, Module module) {
+ tableEnv.loadModule(moduleName, module);
+ }
+
+ @Override
+ public void unloadModule(String moduleName) {
+ tableEnv.unloadModule(moduleName);
+ }
+
+ @Override
+ public void createTemporarySystemFunction(
+ String name, Class functionClass) {
+ tableEnv.createTemporarySystemFunction(name, functionClass);
+ }
+
+ @Override
+ public void createTemporarySystemFunction(
+ String name, UserDefinedFunction functionInstance) {
+ tableEnv.createTemporarySystemFunction(name, functionInstance);
+ }
+
+ @Override
+ public boolean dropTemporarySystemFunction(String name) {
+ return tableEnv.dropTemporarySystemFunction(name);
+ }
+
+ @Override
+ public void createFunction(String path, Class functionClass) {
+ tableEnv.createFunction(path, functionClass);
+ }
+
+ @Override
+ public void createFunction(
+ String path,
+ Class functionClass,
+ boolean ignoreIfExists) {
+ tableEnv.createFunction(path, functionClass, ignoreIfExists);
+ }
+
+ @Override
+ public boolean dropFunction(String path) {
+ return tableEnv.dropFunction(path);
+ }
+
+ @Override
+ public void createTemporaryFunction(
+ String path, Class functionClass) {
+ tableEnv.createTemporaryFunction(path, functionClass);
+ }
+
+ @Override
+ public void createTemporaryFunction(String path, UserDefinedFunction functionInstance) {
+ tableEnv.createTemporaryFunction(path, functionInstance);
+ }
+
+ @Override
+ public boolean dropTemporaryFunction(String path) {
+ return tableEnv.dropTemporaryFunction(path);
+ }
+
+ @Override
+ public void createTemporaryView(String path, Table view) {
+ tableEnv.createTemporaryView(path, view);
+ }
+
+ @Override
+ public Table from(String path) {
+ return tableEnv.from(path);
+ }
+
+ @Override
+ public String[] listCatalogs() {
+ return tableEnv.listCatalogs();
+ }
+
+ @Override
+ public String[] listModules() {
+ return tableEnv.listModules();
+ }
+
+ @Override
+ public String[] listDatabases() {
+ return tableEnv.listDatabases();
+ }
+
+ @Override
+ public String[] listTables() {
+ return tableEnv.listTables();
+ }
+
+ @Override
+ public String[] listViews() {
+ return tableEnv.listViews();
+ }
+
+ @Override
+ public String[] listTemporaryTables() {
+ return tableEnv.listTemporaryTables();
+ }
+
+ @Override
+ public String[] listTemporaryViews() {
+ return tableEnv.listTemporaryViews();
+ }
+
+ @Override
+ public String[] listUserDefinedFunctions() {
+ return tableEnv.listUserDefinedFunctions();
+ }
+
+ @Override
+ public String[] listFunctions() {
+ return tableEnv.listFunctions();
+ }
+
+ @Override
+ public boolean dropTemporaryTable(String path) {
+ return tableEnv.dropTemporaryTable(path);
+ }
+
+ @Override
+ public boolean dropTemporaryView(String path) {
+ return tableEnv.dropTemporaryView(path);
+ }
+
+ @Override
+ public String explainSql(String statement, ExplainDetail... extraDetails) {
+ return tableEnv.explainSql(statement, extraDetails);
+ }
+
+ @Override
+ public Table sqlQuery(String query) {
+ return tableEnv.sqlQuery(query);
+ }
+
+ @Override
+ public TableResult executeSql(String statement) {
+ return tableEnv.executeSql(statement);
+ }
+
+ @Override
+ public String getCurrentCatalog() {
+ return tableEnv.getCurrentCatalog();
+ }
+
+ @Override
+ public void useCatalog(String catalogName) {
+ tableEnv.useCatalog(catalogName);
+ }
+
+ @Override
+ public String getCurrentDatabase() {
+ return tableEnv.getCurrentDatabase();
+ }
+
+ @Override
+ public void useDatabase(String databaseName) {
+ tableEnv.useDatabase(databaseName);
+ }
+
+ @Override
+ public TableConfig getConfig() {
+ return tableEnv.getConfig();
+ }
+
+ @Override
+ public StatementSet createStatementSet() {
+ return tableEnv.createStatementSet();
+ }
+
+ @Override
+ public String[] getCompletionHints(String statement, int position) {
+ return tableEnv.getCompletionHints(statement, position);
+ }
+
+ /** @deprecated Retained for backward compatibility with legacy Flink Table API. */
+ @Deprecated(since = "2.1.0", forRemoval = false)
+ @Override
+ public Table scan(String... tablePath) {
+ return tableEnv.scan(tablePath);
+ }
+
+ /** @deprecated Retained for backward compatibility with legacy Flink Table API. */
+ @Deprecated(since = "2.1.0", forRemoval = false)
+ @Override
+ public void registerTable(String name, Table table) {
+ tableEnv.registerTable(name, table);
+ }
+
+ /** @deprecated Retained for backward compatibility with legacy Flink Table API. */
+ @Deprecated(since = "2.1.0", forRemoval = false)
+ @Override
+ public void registerFunction(String name, ScalarFunction function) {
+ tableEnv.registerFunction(name, function);
+ }
+}
diff --git a/streampark-flink/streampark-flink-shims/streampark-flink-shims-base-v2/src/main/java/org/apache/streampark/flink/core/SqlCommand.java b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base-v2/src/main/java/org/apache/streampark/flink/core/SqlCommand.java
new file mode 100644
index 0000000000..ddc5f5abf4
--- /dev/null
+++ b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base-v2/src/main/java/org/apache/streampark/flink/core/SqlCommand.java
@@ -0,0 +1,180 @@
+/*
+ * 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.streampark.flink.core;
+
+import org.apache.commons.lang3.StringUtils;
+
+import java.util.Optional;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/** Flink SQL command types. */
+public enum SqlCommand {
+
+ // ---- SELECT Statements -----------------------------------------------------------------
+ SELECT("select", "(SELECT\\s+.+)"),
+
+ // ---- CREATE Statements -----------------------------------------------------------------
+ CREATE_TABLE("create table", "(CREATE\\s+(TEMPORARY\\s+|)TABLE\\s+.+)"),
+ CREATE_CATALOG("create catalog", "(CREATE\\s+CATALOG\\s+.+)"),
+ CREATE_DATABASE("create database", "(CREATE\\s+DATABASE\\s+.+)"),
+ CREATE_VIEW(
+ "create view",
+ "(CREATE\\s+(TEMPORARY\\s+|)VIEW\\s+(IF\\s+NOT\\s+EXISTS\\s+|)(\\S+)\\s+AS\\s+SELECT\\s+.+)"),
+ CREATE_FUNCTION(
+ "create function",
+ "(CREATE\\s+(TEMPORARY\\s+|TEMPORARY\\s+SYSTEM\\s+|)FUNCTION\\s+(IF\\s+NOT\\s+EXISTS\\s+|)(\\S+)\\s+AS\\s+.*)"),
+
+ // ---- DROP Statements -------------------------------------------------------------------
+ DROP_CATALOG("drop catalog", "(DROP\\s+CATALOG\\s+.+)"),
+ DROP_TABLE("drop table", "(DROP\\s+(TEMPORARY\\s+|)TABLE\\s+.+)"),
+ DROP_DATABASE("drop database", "(DROP\\s+DATABASE\\s+.+)"),
+ DROP_VIEW("drop view", "(DROP\\s+(TEMPORARY\\s+|)VIEW\\s+.+)"),
+ DROP_FUNCTION(
+ "drop function", "(DROP\\s+(TEMPORARY\\s+|TEMPORARY\\s+SYSTEM\\s+|)FUNCTION\\s+.+)"),
+
+ // ---- ALTER Statements ------------------------------------------------------------------
+ ALTER_TABLE("alter table", "(ALTER\\s+TABLE\\s+.+)"),
+ ALTER_VIEW("alter view", "(ALTER\\s+VIEW\\s+.+)"),
+ ALTER_DATABASE("alter database", "(ALTER\\s+DATABASE\\s+.+)"),
+ ALTER_FUNCTION(
+ "alter function",
+ "(ALTER\\s+(TEMPORARY\\s+|TEMPORARY\\s+SYSTEM\\s+|)FUNCTION\\s+.+)"),
+
+ // ---- INSERT Statement ------------------------------------------------------------------
+ INSERT("insert", "(INSERT\\s+(INTO|OVERWRITE)\\s+.+)"),
+
+ // ---- DESCRIBE Statement ----------------------------------------------------------------
+ DESC("desc", "(DESC\\s+.+)"),
+ DESCRIBE("describe", "(DESCRIBE\\s+.+)"),
+
+ // ---- EXPLAIN Statement -----------------------------------------------------------------
+ EXPLAIN("explain", "(EXPLAIN\\s+.+)"),
+
+ // ---- USE Statements --------------------------------------------------------------------
+ USE_CATALOG("use catalog", "(USE\\s+CATALOG\\s+.+)"),
+ USE_MODULES("use modules", "(USE\\s+MODULES\\s+.+)"),
+ USE_DATABASE("use database", "(USE\\s+(?!(CATALOG|MODULES)).+)"),
+
+ // ---- SHOW Statements -------------------------------------------------------------------
+ SHOW_CATALOGS("show catalogs", "(SHOW\\s+CATALOGS\\s*)"),
+ SHOW_CURRENT_CATALOG("show current catalog", "(SHOW\\s+CURRENT\\s+CATALOG\\s*)"),
+ SHOW_DATABASES("show databases", "(SHOW\\s+DATABASES\\s*)"),
+ SHOW_CURRENT_DATABASE("show current database", "(SHOW\\s+CURRENT\\s+DATABASE\\s*)"),
+ SHOW_TABLES("show tables", "(SHOW\\s+TABLES.*)"),
+ SHOW_CREATE_TABLE("show create table", "(SHOW\\s+CREATE\\s+TABLE\\s+.+)"),
+ SHOW_COLUMNS("show columns", "(SHOW\\s+COLUMNS\\s+.+)"),
+ SHOW_VIEWS("show views", "(SHOW\\s+VIEWS\\s*)"),
+ SHOW_CREATE_VIEW("show create view", "(SHOW\\s+CREATE\\s+VIEW\\s+.+)"),
+ SHOW_FUNCTIONS("show functions", "(SHOW\\s+(USER\\s+|)FUNCTIONS\\s*)"),
+ SHOW_MODULES("show modules", "(SHOW\\s+(FULL\\s+|)MODULES\\s*)"),
+
+ // ---- LOAD Statements -------------------------------------------------------------------
+ LOAD_MODULE("load module", "(LOAD\\s+MODULE\\s+.+)"),
+
+ // ---- UNLOAD Statements -----------------------------------------------------------------
+ UNLOAD_MODULE("unload module", "(UNLOAD\\s+MODULE\\s+.+)"),
+
+ // ---- SET Statements --------------------------------------------------------------------
+ SET(
+ "set",
+ "SET(\\s+(\\S+)\\s*=(.*))?",
+ groups -> {
+ if (groups.length < 3) {
+ return Optional.empty();
+ }
+ if (groups[0] == null) {
+ return Optional.of(new String[]{cleanUp(groups[0])});
+ }
+ return Optional.of(new String[]{cleanUp(groups[1]), cleanUp(groups[2])});
+ }),
+
+ // ---- RESET Statements ------------------------------------------------------------------
+ RESET("reset", "RESET\\s+'(.*)'"),
+ RESET_ALL("reset all", "RESET", groups -> Optional.of(new String[]{"ALL"})),
+
+ // ---- INSERT SET Statements -------------------------------------------------------------
+ /** @deprecated SQL Client syntax; not supported on this platform. */
+ @Deprecated(since = "2.1.0", forRemoval = false)
+ BEGIN_STATEMENT_SET(
+ "begin statement set", "BEGIN\\s+STATEMENT\\s+SET", SqlCommandConverters.NO_OPERANDS),
+ /** @deprecated SQL Client syntax; not supported on this platform. */
+ @Deprecated(since = "2.1.0", forRemoval = false)
+ END_STATEMENT_SET("end statement set", "END", SqlCommandConverters.NO_OPERANDS),
+
+ // Since: 2.1.2 for flink 1.18
+ DELETE("delete", "(DELETE\\s+FROM\\s+.+)"),
+ UPDATE("update", "(UPDATE\\s+.+)");
+
+ private static final int PATTERN_FLAGS = Pattern.CASE_INSENSITIVE | Pattern.DOTALL;
+
+ private final String name;
+ private final String regex;
+ private final SqlCommandConverter converter;
+ private Matcher matcher;
+
+ SqlCommand(String name, String regex) {
+ this(name, regex, SqlCommandConverters.DEFAULT);
+ }
+
+ SqlCommand(String name, String regex, SqlCommandConverter converter) {
+ this.name = name;
+ this.regex = regex;
+ this.converter = converter;
+ }
+
+ /** Command label (e.g. {@code "select"}, {@code "create table"}). */
+ public String getName() {
+ return name;
+ }
+
+ public String getRegex() {
+ return regex;
+ }
+
+ public SqlCommandConverter getConverter() {
+ return converter;
+ }
+
+ public Matcher getMatcher() {
+ return matcher;
+ }
+
+ public boolean matches(String input) {
+ if (StringUtils.isBlank(regex)) {
+ return false;
+ }
+ Pattern pattern = Pattern.compile(regex, PATTERN_FLAGS);
+ matcher = pattern.matcher(input);
+ return matcher.matches();
+ }
+
+ /** Resolve the first matching command for the given statement. */
+ public static SqlCommand get(String stmt) {
+ for (SqlCommand command : values()) {
+ if (command.matches(stmt)) {
+ return command;
+ }
+ }
+ return null;
+ }
+
+ static String cleanUp(String sql) {
+ return sql.trim().replaceAll("^(['\"])|(['\"])$", "");
+ }
+}
diff --git a/streampark-flink/streampark-flink-shims/streampark-flink-shims-base-v2/src/main/java/org/apache/streampark/flink/core/StreamEnvConfigFunction.java b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base-v2/src/main/java/org/apache/streampark/flink/core/StreamEnvConfigFunction.java
new file mode 100644
index 0000000000..d1ed71bb32
--- /dev/null
+++ b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base-v2/src/main/java/org/apache/streampark/flink/core/StreamEnvConfigFunction.java
@@ -0,0 +1,34 @@
+/*
+ * 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.streampark.flink.core;
+
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.util.ParameterTool;
+
+@FunctionalInterface
+public interface StreamEnvConfigFunction {
+
+ /**
+ * When used to initialize StreamExecutionEnvironment, it can be used to implement this function
+ * and customize the parameters to be set...
+ *
+ * @param environment
+ * @param parameterTool
+ */
+ void configuration(StreamExecutionEnvironment environment, ParameterTool parameterTool);
+}
diff --git a/streampark-flink/streampark-flink-shims/streampark-flink-shims_flink-1.12/src/main/java/org/apache/streampark/flink/core/FlinkClusterClient.java b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base-v2/src/main/java/org/apache/streampark/flink/core/TableEnvConfigFunction.java
similarity index 65%
rename from streampark-flink/streampark-flink-shims/streampark-flink-shims_flink-1.12/src/main/java/org/apache/streampark/flink/core/FlinkClusterClient.java
rename to streampark-flink/streampark-flink-shims/streampark-flink-shims-base-v2/src/main/java/org/apache/streampark/flink/core/TableEnvConfigFunction.java
index 4914c46e91..0e57b74b5c 100644
--- a/streampark-flink/streampark-flink-shims/streampark-flink-shims_flink-1.12/src/main/java/org/apache/streampark/flink/core/FlinkClusterClient.java
+++ b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base-v2/src/main/java/org/apache/streampark/flink/core/TableEnvConfigFunction.java
@@ -17,11 +17,18 @@
package org.apache.streampark.flink.core;
-import org.apache.flink.client.program.ClusterClient;
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.util.ParameterTool;
-public class FlinkClusterClient extends FlinkClientTrait {
+@FunctionalInterface
+public interface TableEnvConfigFunction {
- public FlinkClusterClient(ClusterClient clusterClient) {
- super(clusterClient);
- }
+ /**
+ * When used to initialize the TableEnvironment, it can be used to implement this function and
+ * customize the parameters to be set...
+ *
+ * @param tableConfig
+ * @param parameterTool
+ */
+ void configuration(TableConfig tableConfig, ParameterTool parameterTool);
}
diff --git a/streampark-flink/streampark-flink-shims/streampark-flink-shims_flink-1.15/src/main/java/org/apache/streampark/flink/core/TableContext.java b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base-v2/src/main/java/org/apache/streampark/flink/core/conf/FlinkConfiguration.java
similarity index 52%
rename from streampark-flink/streampark-flink-shims/streampark-flink-shims_flink-1.15/src/main/java/org/apache/streampark/flink/core/TableContext.java
rename to streampark-flink/streampark-flink-shims/streampark-flink-shims-base-v2/src/main/java/org/apache/streampark/flink/core/conf/FlinkConfiguration.java
index 95199e4983..c55457631b 100644
--- a/streampark-flink/streampark-flink-shims/streampark-flink-shims_flink-1.15/src/main/java/org/apache/streampark/flink/core/TableContext.java
+++ b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base-v2/src/main/java/org/apache/streampark/flink/core/conf/FlinkConfiguration.java
@@ -15,29 +15,26 @@
* limitations under the License.
*/
-package org.apache.streampark.flink.core;
+package org.apache.streampark.flink.core.conf;
-import org.apache.flink.api.common.JobExecutionResult;
-import org.apache.flink.api.java.utils.ParameterTool;
-import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.util.ParameterTool;
-import scala.Tuple2;
+/** Flink runtime configuration holder. */
+public class FlinkConfiguration {
-public class TableContext extends FlinkTableTrait {
+ public final ParameterTool parameter;
+ public final Configuration envConfig;
+ public final Configuration tableConfig;
- public TableContext(ParameterTool parameter, TableEnvironment tableEnv) {
- super(parameter, tableEnv);
+ public FlinkConfiguration(
+ ParameterTool parameter, Configuration envConfig, Configuration tableConfig) {
+ this.parameter = parameter;
+ this.envConfig = envConfig;
+ this.tableConfig = tableConfig;
}
- public TableContext(Tuple2 args) {
- this(args._1(), args._2());
- }
-
- public TableContext(TableEnvConfig args) {
- this(FlinkTableInitializer.initialize(args));
- }
-
- public JobExecutionResult execute(String jobName) {
- return printStartupLogo(jobName);
+ public FlinkConfiguration withParameter(ParameterTool parameter) {
+ return new FlinkConfiguration(parameter, envConfig, tableConfig);
}
}
diff --git a/streampark-flink/streampark-flink-shims/streampark-flink-shims-base-v2/src/main/java/org/apache/streampark/flink/util/FlinkUtils.java b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base-v2/src/main/java/org/apache/streampark/flink/util/FlinkUtils.java
new file mode 100644
index 0000000000..441dbf3d7b
--- /dev/null
+++ b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base-v2/src/main/java/org/apache/streampark/flink/util/FlinkUtils.java
@@ -0,0 +1,79 @@
+/*
+ * 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.streampark.flink.util;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.configuration.CheckpointingOptions;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import org.apache.flink.util.TimeUtils;
+
+import java.io.File;
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/** Flink utility methods. */
+public final class FlinkUtils {
+
+ private FlinkUtils() {
+ }
+
+ public static ListState getUnionListState(
+ FunctionInitializationContext context,
+ String descriptorName,
+ TypeInformation typeInformation) {
+ try {
+ return context.getOperatorStateStore()
+ .getUnionListState(
+ new ListStateDescriptor<>(
+ descriptorName, typeInformation.getTypeClass()));
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ public static String getFlinkDistJar(String flinkHome) {
+ String[] jars =
+ new File(flinkHome + "/lib")
+ .list((dir, name) -> name.matches("flink-dist.*\\.jar"));
+ if (jars == null || jars.length == 0) {
+ throw new IllegalArgumentException(
+ "[StreamPark] can no found flink-dist jar in " + flinkHome + "/lib");
+ }
+ if (jars.length == 1) {
+ return flinkHome + "/lib/" + jars[0];
+ }
+ throw new IllegalArgumentException(
+ "[StreamPark] found multiple flink-dist jar in "
+ + flinkHome
+ + "/lib,["
+ + Arrays.stream(jars).collect(Collectors.joining(","))
+ + "]");
+ }
+
+ public static boolean isCheckpointEnabled(Map map) {
+ Duration checkpointInterval =
+ TimeUtils.parseDuration(
+ map.getOrDefault(
+ CheckpointingOptions.CHECKPOINTING_INTERVAL.key(), "0ms"));
+ return checkpointInterval.toMillis() > 0;
+ }
+}
diff --git a/streampark-flink/streampark-flink-shims/streampark-flink-shims-base-v2/src/test/java/org/apache/streampark/RegExpTest.java b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base-v2/src/test/java/org/apache/streampark/RegExpTest.java
new file mode 100644
index 0000000000..dc7e648e1d
--- /dev/null
+++ b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base-v2/src/test/java/org/apache/streampark/RegExpTest.java
@@ -0,0 +1,92 @@
+/*
+ * 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.streampark;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/** some simple tests */
+class RegExpTest {
+
+ /**
+ * Case insensitive, matches everything as one line, . matches newlines, note: \s matches any
+ * whitespace character, including newlines
+ */
+ public static final int DEFAULT_PATTERN_FLAGS = Pattern.CASE_INSENSITIVE | Pattern.DOTALL;
+
+ /**
+ * CREATE CATALOG catalog_name WITH (key1=val1, key2=val2, ...)
+ * Example：create catalog hive_catalog with('name' = 'my_hive', 'conf' = '/home/hive/conf')
+ */
+ private static final Pattern CREATE_HIVE_CATALOG =
+ Pattern.compile("CREATE\\s+CATALOG\\s+\\S[\\s\\S]*", DEFAULT_PATTERN_FLAGS);
+
+ @Test
+ void testCreateHiveCatalog() {
+ String str = "create catalog hive with (\n"
+ + " 'type' = 'hive',\n"
+ + " 'hadoop-conf-dir' = 'D:\\IDEAWorkspace\\work\\baishan\\log\\data-max\\src\\main\\resources',\n"
+ + " 'hive-conf-dir' = 'D:\\IDEAWorkspace\\work\\baishan\\log\\data-max\\src\\main\\resources'\n"
+ + ")";
+ Matcher matcher = CREATE_HIVE_CATALOG.matcher(str);
+ System.out.println(matcher.matches());
+ }
+
+ /**
+ * CREATE [TEMPORARY|TEMPORARY SYSTEM] FUNCTION [IF NOT EXISTS]
+ * [catalog_name.][db_name.]function_name AS identifier [LANGUAGE JAVA|SCALA|PYTHON]
+ * Example：create function test_fun as com.flink.testFun
+ */
+ private static final Pattern CREATE_FUNCTION =
+ Pattern.compile(
+ "CREATE\\s+(?:TEMPORARY\\s+(?:SYSTEM\\s+)?)?FUNCTION\\s+(?:IF NOT EXISTS\\s+)?"
+ + "([A-Za-z][A-Za-z\\d.\\-_]*)\\s+AS\\s+'([^']+)'\\s+LANGUAGE\\s+(JAVA|SCALA|PYTHON)",
+ DEFAULT_PATTERN_FLAGS);
+
+ @Test
+ void testCreateFunction() {
+ String str =
+ "create function if not exists hive.get_json_value as 'com.flink.function.JsonValueFunction' language java";
+ Matcher matcher = CREATE_FUNCTION.matcher(str);
+ System.out.println(matcher.matches());
+ }
+
+ /** USE [catalog_name.]database_name */
+ private static final Pattern USE_DATABASE =
+ Pattern.compile("USE\\s+(?!(?:CATALOG|MODULES)\\b)\\S[\\s\\S]*", DEFAULT_PATTERN_FLAGS);
+
+ @Test
+ void testUseDatabase() {
+ String str = "use modul.a ";
+ Matcher matcher = USE_DATABASE.matcher(str);
+ System.out.println(matcher.matches());
+ }
+
+ /** SHOW [USER] FUNCTIONS */
+ private static final Pattern SHOW_FUNCTIONS =
+ Pattern.compile("SHOW\\s+(?:USER\\s+)?FUNCTIONS\\b", DEFAULT_PATTERN_FLAGS);
+
+ @Test
+ void testShowFunction() {
+ String str = "show user functions";
+ Matcher matcher = SHOW_FUNCTIONS.matcher(str);
+ System.out.println(matcher.matches());
+ }
+}
diff --git a/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/pom.xml b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/pom.xml
index c5647870ae..c7734c6d8e 100644
--- a/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/pom.xml
+++ b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/pom.xml
@@ -25,14 +25,14 @@
${revision}
- streampark-flink-shims-base_${scala.binary.version}
+ streampark-flink-shims-base
StreamPark : Flink Shims Base
org.apache.streampark
- streampark-common_${scala.binary.version}
+ streampark-common
@@ -43,13 +43,6 @@
provided