This is an automated email from the ASF dual-hosted git repository. libenchao pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push: new 06688f345f6 [FLINK-31687][jdbc-driver] Get rid of flink-core for jdbc driver 06688f345f6 is described below commit 06688f345f6793a8964ec00002175f44cda13c33 Author: Shammon FY <zjur...@gmail.com> AuthorDate: Sat May 6 09:51:44 2023 +0800 [FLINK-31687][jdbc-driver] Get rid of flink-core for jdbc driver Close apache/flink#22533 --- .../client/cli/parser/SqlClientSyntaxHighlighter.java | 8 ++++---- .../apache/flink/table/client/gateway/Executor.java | 8 ++++++++ .../flink/table/client/gateway/ExecutorImpl.java | 11 ++++++++++- .../apache/flink/table/client/cli/CliClientTest.java | 9 ++++++++- .../table/gateway/service/context/DefaultContext.java | 5 +++++ flink-table/flink-sql-jdbc-driver-bundle/pom.xml | 9 --------- .../org/apache/flink/table/jdbc/FlinkConnection.java | 19 ++++++------------- 7 files changed, 41 insertions(+), 28 deletions(-) diff --git a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/parser/SqlClientSyntaxHighlighter.java b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/parser/SqlClientSyntaxHighlighter.java index bdcfa5a707d..366cae3f57e 100644 --- a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/parser/SqlClientSyntaxHighlighter.java +++ b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/parser/SqlClientSyntaxHighlighter.java @@ -18,6 +18,7 @@ package org.apache.flink.table.client.cli.parser; +import org.apache.flink.configuration.ReadableConfig; import org.apache.flink.sql.parser.impl.FlinkSqlParserImplConstants; import org.apache.flink.table.api.SqlDialect; import org.apache.flink.table.api.config.TableConfigOptions; @@ -57,16 +58,15 @@ public class SqlClientSyntaxHighlighter extends DefaultHighlighter { @Override public AttributedString highlight(LineReader reader, String buffer) { + ReadableConfig configuration = executor.getSessionConfig(); final SyntaxHighlightStyle.BuiltInStyle style = SyntaxHighlightStyle.BuiltInStyle.fromString( - executor.getSessionConfig() - .get(SqlClientOptions.DISPLAY_DEFAULT_COLOR_SCHEMA)); + configuration.get(SqlClientOptions.DISPLAY_DEFAULT_COLOR_SCHEMA)); if (style == SyntaxHighlightStyle.BuiltInStyle.DEFAULT) { return super.highlight(reader, buffer); } - final String dialectName = - executor.getSessionConfig().get(TableConfigOptions.TABLE_SQL_DIALECT); + final String dialectName = configuration.get(TableConfigOptions.TABLE_SQL_DIALECT); final SqlDialect dialect = SqlDialect.HIVE.name().equalsIgnoreCase(dialectName) ? SqlDialect.HIVE diff --git a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/Executor.java b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/Executor.java index 8414789c993..b636d326560 100644 --- a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/Executor.java +++ b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/Executor.java @@ -25,6 +25,7 @@ import java.io.Closeable; import java.net.InetSocketAddress; import java.net.URL; import java.util.List; +import java.util.Map; /** A gateway for communicating with Flink and other external systems. */ public interface Executor extends Closeable { @@ -53,6 +54,13 @@ public interface Executor extends Closeable { */ ReadableConfig getSessionConfig(); + /** + * Get the map configuration of the session. + * + * @return the map session configuration. + */ + Map<String, String> getSessionConfigMap(); + /** * Execute statement. * diff --git a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/ExecutorImpl.java b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/ExecutorImpl.java index fcdee3c779d..4ec7eb3e8eb 100644 --- a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/ExecutorImpl.java +++ b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/ExecutorImpl.java @@ -83,6 +83,7 @@ import java.net.URL; import java.util.Collections; import java.util.Iterator; import java.util.List; +import java.util.Map; import java.util.UUID; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; @@ -206,6 +207,14 @@ public class ExecutorImpl implements Executor { } public ReadableConfig getSessionConfig() { + try { + return Configuration.fromMap(getSessionConfigMap()); + } catch (Exception e) { + throw new SqlExecutionException("Failed to get the get session config.", e); + } + } + + public Map<String, String> getSessionConfigMap() { try { GetSessionConfigResponseBody response = getResponse( @@ -213,7 +222,7 @@ public class ExecutorImpl implements Executor { GetSessionConfigHeaders.getInstance(), new SessionMessageParameters(sessionHandle), EmptyRequestBody.getInstance())); - return Configuration.fromMap(response.getProperties()); + return response.getProperties(); } catch (Exception e) { throw new SqlExecutionException("Failed to get the get session config.", e); } diff --git a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliClientTest.java b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliClientTest.java index 9ce1fc1c808..bbfdc64b094 100644 --- a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliClientTest.java +++ b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliClientTest.java @@ -20,6 +20,7 @@ package org.apache.flink.table.client.cli; import org.apache.flink.api.common.JobID; import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.ReadableConfig; import org.apache.flink.runtime.testutils.CommonTestUtils; import org.apache.flink.table.api.DataTypes; import org.apache.flink.table.api.ResultKind; @@ -61,6 +62,7 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.List; +import java.util.Map; import static org.apache.flink.table.api.config.TableConfigOptions.TABLE_DML_SYNC; import static org.apache.flink.table.api.internal.StaticResultProvider.SIMPLE_ROW_DATA_TO_STRING_CONVERTER; @@ -406,10 +408,15 @@ class CliClientTest { public void configureSession(String statement) {} @Override - public Configuration getSessionConfig() { + public ReadableConfig getSessionConfig() { return configuration; } + @Override + public Map<String, String> getSessionConfigMap() { + return configuration.toMap(); + } + @Override public StatementResult executeStatement(String statement) { receivedStatement = statement; diff --git a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/context/DefaultContext.java b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/context/DefaultContext.java index a48699d737f..b914ad4e7f1 100644 --- a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/context/DefaultContext.java +++ b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/context/DefaultContext.java @@ -42,6 +42,7 @@ import java.nio.file.Paths; import java.util.ArrayList; import java.util.Collections; import java.util.List; +import java.util.Map; import java.util.stream.Collectors; import java.util.stream.Stream; @@ -52,6 +53,10 @@ public class DefaultContext { private final Configuration flinkConfig; private final List<URL> dependencies; + public DefaultContext(Map<String, String> flinkConfig, List<URL> dependencies) { + this(Configuration.fromMap(flinkConfig), dependencies); + } + public DefaultContext(Configuration flinkConfig, List<URL> dependencies) { this.flinkConfig = flinkConfig; this.dependencies = dependencies; diff --git a/flink-table/flink-sql-jdbc-driver-bundle/pom.xml b/flink-table/flink-sql-jdbc-driver-bundle/pom.xml index 7bdc62b30f0..cafbe51032d 100644 --- a/flink-table/flink-sql-jdbc-driver-bundle/pom.xml +++ b/flink-table/flink-sql-jdbc-driver-bundle/pom.xml @@ -83,14 +83,6 @@ <build> <plugins> - <plugin> - <groupId>io.github.zentol.japicmp</groupId> - <artifactId>japicmp-maven-plugin</artifactId> - <configuration> - <!-- TODO this should be removed after get rid of flink core in issue https://issues.apache.org/jira/browse/FLINK-31687. --> - <skip>true</skip> - </configuration> - </plugin> <!-- Build flink-sql-gateway jar --> <plugin> <groupId>org.apache.maven.plugins</groupId> @@ -112,7 +104,6 @@ <include>org.apache.flink:flink-sql-gateway</include> <include>org.apache.flink:flink-table-common</include> <include>org.apache.flink:flink-annotations</include> - <include>org.apache.flink:flink-core</include> </includes> </artifactSet> </configuration> diff --git a/flink-table/flink-sql-jdbc-driver/src/main/java/org/apache/flink/table/jdbc/FlinkConnection.java b/flink-table/flink-sql-jdbc-driver/src/main/java/org/apache/flink/table/jdbc/FlinkConnection.java index 12203f94541..8c0f1a237c6 100644 --- a/flink-table/flink-sql-jdbc-driver/src/main/java/org/apache/flink/table/jdbc/FlinkConnection.java +++ b/flink-table/flink-sql-jdbc-driver/src/main/java/org/apache/flink/table/jdbc/FlinkConnection.java @@ -19,7 +19,6 @@ package org.apache.flink.table.jdbc; import org.apache.flink.annotation.VisibleForTesting; -import org.apache.flink.configuration.Configuration; import org.apache.flink.table.client.gateway.Executor; import org.apache.flink.table.client.gateway.StatementResult; import org.apache.flink.table.gateway.service.context.DefaultContext; @@ -35,6 +34,7 @@ import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.Properties; import java.util.UUID; @@ -53,13 +53,10 @@ public class FlinkConnection extends BaseConnection { public FlinkConnection(DriverUri driverUri) { this.url = driverUri.getURL(); this.statements = new ArrayList<>(); - // TODO Support default context from map to get gid of flink core for jdbc driver in - // https://issues.apache.org/jira/browse/FLINK-31687. this.executor = Executor.create( new DefaultContext( - Configuration.fromMap( - DriverUtils.fromProperties(driverUri.getProperties())), + DriverUtils.fromProperties(driverUri.getProperties()), Collections.emptyList()), driverUri.getAddress(), UUID.randomUUID().toString()); @@ -165,20 +162,16 @@ public class FlinkConnection extends BaseConnection { @Override public String getClientInfo(String name) throws SQLException { ensureOpen(); - // TODO Executor should return Map<String, String> here to get rid of flink core for jdbc - // driver in https://issues.apache.org/jira/browse/FLINK-31687. - Configuration configuration = (Configuration) executor.getSessionConfig(); - return configuration.toMap().get(name); + Map<String, String> configuration = executor.getSessionConfigMap(); + return configuration.get(name); } @Override public Properties getClientInfo() throws SQLException { ensureOpen(); Properties properties = new Properties(); - // TODO Executor should return Map<String, String> here to get rid of flink core for jdbc - // driver in https://issues.apache.org/jira/browse/FLINK-31687. - Configuration configuration = (Configuration) executor.getSessionConfig(); - configuration.toMap().forEach(properties::setProperty); + Map<String, String> configuration = executor.getSessionConfigMap(); + configuration.forEach(properties::setProperty); return properties; }