This is an automated email from the ASF dual-hosted git repository.
lihaosky 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 42290314753 [FLINK-38262][table] show create connection operation
(#28791)
42290314753 is described below
commit 42290314753281773b13ce504038d6d8493b37a0
Author: Shekhar Rajak <[email protected]>
AuthorDate: Wed Sep 16 02:22:30 2026 +0530
[FLINK-38262][table] show create connection operation (#28791)
* [FLINK-38262][table] Add SHOW CREATE CONNECTION operation
Generated-by: Codex (GPT-5)
* Update
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java
Co-authored-by: Ramin Gharib <[email protected]>
* Update
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java
Co-authored-by: Ramin Gharib <[email protected]>
* [FLINK-38262][table] Add SHOW CREATE CONNECTION util tests
* [FLINK-38262][table] Hide connection secret metadata
* [FLINK-38262][table] Strengthen connection show create test
* [FLINK-38262][table] Fix secret-only connection show create
* [FLINK-38262][table] Reuse connection secret reference key
* [FLINK-38262][table] Reuse connection lookup
* [FLINK-38262][table] Reuse comment extraction
* [FLINK-38262][table] Test permanent connection show create
---------
Co-authored-by: Ramin Gharib <[email protected]>
---
.../flink/table/api/internal/ShowCreateUtil.java | 64 ++++++++++++---
.../apache/flink/table/catalog/CatalogManager.java | 12 +++
.../table/catalog/ContextResolvedConnection.java | 54 +++++++++++++
.../operations/ShowCreateConnectionOperation.java | 74 +++++++++++++++++
.../table/api/internal/ShowCreateUtilTest.java | 83 +++++++++++++++++++
.../flink/table/catalog/CatalogManagerTest.java | 42 ++++++++++
.../table/factories/DefaultConnectionFactory.java | 2 +-
.../operations/converters/SqlNodeConverters.java | 1 +
.../SqlShowCreateConnectionConverter.java | 55 +++++++++++++
.../SqlConnectionOperationConverterTest.java | 48 +++++++++++
.../runtime/batch/sql/CreateConnectionITCase.java | 93 ++++++++++++++++++++++
11 files changed, 515 insertions(+), 13 deletions(-)
diff --git
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/ShowCreateUtil.java
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/ShowCreateUtil.java
index 0e690c4d0cb..d8c0fa9306d 100644
---
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/ShowCreateUtil.java
+++
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/ShowCreateUtil.java
@@ -23,6 +23,7 @@ import org.apache.flink.configuration.GlobalConfiguration;
import org.apache.flink.table.api.TableException;
import org.apache.flink.table.catalog.CatalogBaseTable;
import org.apache.flink.table.catalog.CatalogBaseTable.TableKind;
+import org.apache.flink.table.catalog.CatalogConnection;
import org.apache.flink.table.catalog.CatalogDescriptor;
import org.apache.flink.table.catalog.CatalogView;
import org.apache.flink.table.catalog.Column;
@@ -39,6 +40,8 @@ import org.apache.flink.table.catalog.StartMode;
import org.apache.flink.table.catalog.TableDistribution;
import org.apache.flink.table.catalog.UniqueConstraint;
import org.apache.flink.table.expressions.SqlFactory;
+import org.apache.flink.table.factories.DefaultConnectionFactory;
+import org.apache.flink.table.factories.FactoryUtil;
import org.apache.flink.table.utils.EncodingUtils;
import org.apache.commons.lang3.StringUtils;
@@ -84,7 +87,7 @@ public class ShowCreateUtil {
extractFormattedColumns(model.getResolvedOutputSchema())
.ifPresent(
c -> sb.append(String.format("OUTPUT (%s)%s", c,
System.lineSeparator())));
- extractComment(model)
+ extractComment(model.getComment())
.ifPresent(c ->
sb.append(formatComment(c)).append(System.lineSeparator()));
extractFormattedOptions(model.getOptions(), PRINT_INDENT,
additionalSensitiveKeys)
.ifPresent(
@@ -99,6 +102,41 @@ public class ShowCreateUtil {
return sb.toString();
}
+ public static String buildShowCreateConnectionRow(
+ CatalogConnection connection,
+ ObjectIdentifier connectionIdentifier,
+ boolean isTemporary,
+ List<String> additionalSensitiveKeys) {
+ StringBuilder sb =
+ new StringBuilder()
+ .append(
+ buildCreateFormattedPrefix(
+ "CONNECTION",
+ isTemporary,
+ connectionIdentifier,
+ false,
+ false));
+ extractComment(connection.getComment())
+ .ifPresent(c -> sb.append(formatComment(c)).append("\n"));
+ final Map<String, String> connectionOptions =
+ withoutConnectionInternalOptions(connection.getOptions());
+ extractFormattedOptions(
+ connectionOptions.isEmpty()
+ && connection
+ .getOptions()
+ .containsKey(
+
DefaultConnectionFactory
+
.SECRET_REFERENCE_KEY)
+ ? Map.of(
+ FactoryUtil.CONNECTION_TYPE.key(),
+
FactoryUtil.CONNECTION_TYPE.defaultValue())
+ : connectionOptions,
+ PRINT_INDENT,
+ additionalSensitiveKeys)
+ .ifPresent(v -> sb.append("WITH
(\n").append(v).append("\n)\n"));
+ return sb.toString();
+ }
+
public static String buildShowCreateTableRow(
ResolvedCatalogBaseTable<?> table,
ObjectIdentifier tableIdentifier,
@@ -117,7 +155,7 @@ public class ShowCreateUtil {
extractFormattedPrimaryKey(table, PRINT_INDENT)
.ifPresent(pk -> sb.append(",\n").append(pk));
sb.append("\n)\n");
- extractComment(table).ifPresent(c ->
sb.append(formatComment(c)).append("\n"));
+ extractComment(table.getComment()).ifPresent(c ->
sb.append(formatComment(c)).append("\n"));
extractFormattedDistributedInfo((ResolvedCatalogTable) table)
.ifPresent(d -> sb.append(d).append("\n"));
extractFormattedPartitionedInfo((ResolvedCatalogTable) table)
@@ -233,7 +271,7 @@ public class ShowCreateUtil {
extractFormattedPrimaryKey(table, PRINT_INDENT)
.ifPresent(pk -> sb.append(",\n").append(pk));
sb.append("\n)\n");
- extractComment(table).ifPresent(c ->
sb.append(formatComment(c)).append("\n"));
+ extractComment(table.getComment()).ifPresent(c ->
sb.append(formatComment(c)).append("\n"));
table.getDistribution()
.map(TableDistribution::toString)
.ifPresent(d -> sb.append(d).append("\n"));
@@ -270,7 +308,7 @@ public class ShowCreateUtil {
buildCreateFormattedPrefix(
"VIEW", isTemporary, viewIdentifier,
false, true));
sb.append(extractFormattedColumnNames(view,
PRINT_INDENT)).append("\n)\n");
- extractComment(view).ifPresent(c ->
sb.append(formatComment(c)).append("\n"));
+ extractComment(view.getComment()).ifPresent(c ->
sb.append(formatComment(c)).append("\n"));
sb.append("AS ").append(((CatalogView)
origin).getExpandedQuery()).append("\n");
return sb.toString();
@@ -394,16 +432,18 @@ public class ShowCreateUtil {
return String.format("PARTITIONED BY (%s)\n", partitionedByColumns);
}
- static Optional<String> extractComment(ResolvedCatalogBaseTable<?> table) {
- return StringUtils.isEmpty(table.getComment())
- ? Optional.empty()
- : Optional.of(table.getComment());
+ private static Optional<String> extractComment(String comment) {
+ return StringUtils.isEmpty(comment) ? Optional.empty() :
Optional.of(comment);
}
- static Optional<String> extractComment(ResolvedCatalogModel model) {
- return StringUtils.isEmpty(model.getComment())
- ? Optional.empty()
- : Optional.of(model.getComment());
+ private static Map<String, String> withoutConnectionInternalOptions(
+ Map<String, String> options) {
+ return options.entrySet().stream()
+ .filter(
+ entry ->
+
!DefaultConnectionFactory.SECRET_REFERENCE_KEY.equals(
+ entry.getKey()))
+ .collect(Collectors.toMap(Map.Entry::getKey,
Map.Entry::getValue));
}
static Optional<String>
extractFormattedDistributedInfo(ResolvedCatalogTable catalogTable) {
diff --git
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java
index fa9db8cb50d..9a306d4e0bc 100644
---
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java
+++
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java
@@ -1924,6 +1924,18 @@ public final class CatalogManager implements
CatalogRegistry, AutoCloseable {
}
}
+ /** Get a connection from the catalog with contextual metadata. */
+ public Optional<ContextResolvedConnection> getResolvedConnection(
+ ObjectIdentifier objectIdentifier) {
+ return getConnection(objectIdentifier)
+ .map(
+ connection ->
+ ContextResolvedConnection.of(
+ objectIdentifier,
+ connection,
+
temporaryConnections.containsKey(objectIdentifier)));
+ }
+
/**
* List all connections in the given catalog and database.
*
diff --git
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/ContextResolvedConnection.java
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/ContextResolvedConnection.java
new file mode 100644
index 00000000000..5df73f53d35
--- /dev/null
+++
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/ContextResolvedConnection.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.catalog;
+
+import org.apache.flink.annotation.Internal;
+
+/** Contextual metadata for a resolved connection. */
+@Internal
+public final class ContextResolvedConnection {
+
+ private final ObjectIdentifier identifier;
+ private final CatalogConnection connection;
+ private final boolean temporary;
+
+ private ContextResolvedConnection(
+ ObjectIdentifier identifier, CatalogConnection connection, boolean
temporary) {
+ this.identifier = identifier;
+ this.connection = connection;
+ this.temporary = temporary;
+ }
+
+ public static ContextResolvedConnection of(
+ ObjectIdentifier identifier, CatalogConnection connection, boolean
temporary) {
+ return new ContextResolvedConnection(identifier, connection,
temporary);
+ }
+
+ public ObjectIdentifier getIdentifier() {
+ return identifier;
+ }
+
+ public CatalogConnection getConnection() {
+ return connection;
+ }
+
+ public boolean isTemporary() {
+ return temporary;
+ }
+}
diff --git
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ShowCreateConnectionOperation.java
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ShowCreateConnectionOperation.java
new file mode 100644
index 00000000000..f3f25cd0b46
--- /dev/null
+++
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ShowCreateConnectionOperation.java
@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.operations;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.SecurityOptions;
+import org.apache.flink.table.api.internal.ShowCreateUtil;
+import org.apache.flink.table.api.internal.TableResultInternal;
+import org.apache.flink.table.catalog.CatalogConnection;
+import org.apache.flink.table.catalog.ObjectIdentifier;
+
+import static
org.apache.flink.table.api.internal.TableResultUtils.buildStringArrayResult;
+
+/** Operation to describe a SHOW CREATE CONNECTION statement. */
+@Internal
+public class ShowCreateConnectionOperation implements ShowOperation {
+
+ private final ObjectIdentifier connectionIdentifier;
+ private final CatalogConnection connection;
+ private final boolean isTemporary;
+
+ public ShowCreateConnectionOperation(
+ ObjectIdentifier connectionIdentifier,
+ CatalogConnection connection,
+ boolean isTemporary) {
+ this.connectionIdentifier = connectionIdentifier;
+ this.connection = connection;
+ this.isTemporary = isTemporary;
+ }
+
+ public ObjectIdentifier getConnectionIdentifier() {
+ return connectionIdentifier;
+ }
+
+ public CatalogConnection getConnection() {
+ return connection;
+ }
+
+ public boolean isTemporary() {
+ return isTemporary;
+ }
+
+ @Override
+ public String asSummaryString() {
+ return String.format("SHOW CREATE CONNECTION %s",
connectionIdentifier.asSummaryString());
+ }
+
+ @Override
+ public TableResultInternal execute(Context ctx) {
+ String resultRow =
+ ShowCreateUtil.buildShowCreateConnectionRow(
+ connection,
+ connectionIdentifier,
+ isTemporary,
+
ctx.getTableConfig().get(SecurityOptions.ADDITIONAL_SENSITIVE_KEYS));
+ return buildStringArrayResult("result", new String[] {resultRow});
+ }
+}
diff --git
a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/api/internal/ShowCreateUtilTest.java
b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/api/internal/ShowCreateUtilTest.java
index 9a4391973c8..9b0ce1d7b4c 100644
---
a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/api/internal/ShowCreateUtilTest.java
+++
b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/api/internal/ShowCreateUtilTest.java
@@ -22,6 +22,7 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogConnection;
import org.apache.flink.table.catalog.CatalogDescriptor;
import org.apache.flink.table.catalog.CatalogMaterializedTable;
import
org.apache.flink.table.catalog.CatalogMaterializedTable.LogicalRefreshMode;
@@ -44,6 +45,7 @@ import org.apache.flink.table.catalog.StartMode.StartModeKind;
import org.apache.flink.table.catalog.TableDistribution;
import org.apache.flink.table.catalog.UniqueConstraint;
import org.apache.flink.table.expressions.DefaultSqlFactory;
+import org.apache.flink.table.factories.DefaultConnectionFactory;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
@@ -69,6 +71,8 @@ class ShowCreateUtilTest {
ObjectIdentifier.of("catalogName", "dbName", "tableName");
private static final ObjectIdentifier VIEW_IDENTIFIER =
ObjectIdentifier.of("catalogName", "dbName", "viewName");
+ private static final ObjectIdentifier CONNECTION_IDENTIFIER =
+ ObjectIdentifier.of("catalogName", "dbName", "connectionName");
private static final ObjectIdentifier MATERIALIZED_TABLE_IDENTIFIER =
ObjectIdentifier.of("catalogName", "dbName",
"materializedTableName");
@@ -227,6 +231,69 @@ class ShowCreateUtilTest {
assertThat(createCatalogString).isEqualTo(expected);
}
+ @ParameterizedTest(name = "{index}: {3}")
+ @MethodSource("argsForShowCreateConnection")
+ void showCreateConnection(
+ CatalogConnection catalogConnection,
+ boolean isTemporary,
+ List<String> additionalSensitiveKeys,
+ String expected) {
+ final String createConnectionString =
+ ShowCreateUtil.buildShowCreateConnectionRow(
+ catalogConnection,
+ CONNECTION_IDENTIFIER,
+ isTemporary,
+ additionalSensitiveKeys);
+ assertThat(createConnectionString).isEqualTo(expected);
+ }
+
+ private static Collection<Arguments> argsForShowCreateConnection() {
+ final Collection<Arguments> argList = new ArrayList<>();
+
+ final Map<String, String> options = new HashMap<>();
+ options.put("url", "jdbc:mysql://localhost:3306/db");
+ options.put("type", "default");
+ addTemporaryAndPermanent(
+ argList,
+ CatalogConnection.of(options, null),
+ List.of(),
+ "CREATE %sCONNECTION `catalogName`.`dbName`.`connectionName`\n"
+ + "WITH (\n"
+ + " 'type' = 'default',\n"
+ + " 'url' = 'jdbc:mysql://localhost:3306/db'\n"
+ + ")\n");
+
+ final Map<String, String> commentedOptions = new HashMap<>();
+ commentedOptions.put("connector", "jdbc");
+ addTemporaryAndPermanent(
+ argList,
+ CatalogConnection.of(commentedOptions, "Connection's comment"),
+ List.of(),
+ "CREATE %sCONNECTION `catalogName`.`dbName`.`connectionName`\n"
+ + "COMMENT 'Connection''s comment'\n"
+ + "WITH (\n"
+ + " 'connector' = 'jdbc'\n"
+ + ")\n");
+
+ final Map<String, String> sensitiveOptions = new HashMap<>();
+ sensitiveOptions.put(DefaultConnectionFactory.SECRET_REFERENCE_KEY,
"secret-id");
+ sensitiveOptions.put("endpoint", "service");
+ sensitiveOptions.put("my.custom.cred", "supersecret");
+ sensitiveOptions.put("password", "topsecret");
+ addTemporaryAndPermanent(
+ argList,
+ CatalogConnection.of(sensitiveOptions, null),
+ List.of("custom.cred"),
+ "CREATE %sCONNECTION `catalogName`.`dbName`.`connectionName`\n"
+ + "WITH (\n"
+ + " 'endpoint' = 'service',\n"
+ + " 'my.custom.cred' = '******',\n"
+ + " 'password' = '******'\n"
+ + ")\n");
+
+ return argList;
+ }
+
private static Collection<Arguments> argsForShowCreateCatalog() {
Collection<Arguments> argList = new ArrayList<>();
Map<String, String> options = new HashMap<>();
@@ -583,6 +650,22 @@ class ShowCreateUtilTest {
argList.add(Arguments.of(catalogBaseTable, true, String.format(sql,
"TEMPORARY ")));
}
+ private static void addTemporaryAndPermanent(
+ Collection<Arguments> argList,
+ CatalogConnection catalogConnection,
+ List<String> additionalSensitiveKeys,
+ String sql) {
+ argList.add(
+ Arguments.of(
+ catalogConnection, false, additionalSensitiveKeys,
String.format(sql, "")));
+ argList.add(
+ Arguments.of(
+ catalogConnection,
+ true,
+ additionalSensitiveKeys,
+ String.format(sql, "TEMPORARY ")));
+ }
+
private static void addCreateAndCreateOrAlter(
Collection<Arguments> argList, CatalogBaseTable catalogBaseTable,
String sql) {
argList.add(Arguments.of(catalogBaseTable, false, String.format(sql,
"CREATE ")));
diff --git
a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/CatalogManagerTest.java
b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/CatalogManagerTest.java
index 51774ffb6f6..544cb3f8fa1 100644
---
a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/CatalogManagerTest.java
+++
b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/CatalogManagerTest.java
@@ -527,6 +527,48 @@ class CatalogManagerTest {
assertThat(dropTemporaryEvent.identifier().getObjectName()).isEqualTo("conn2");
}
+ @Test
+ void testGetResolvedConnection() throws Exception {
+ CatalogManager catalogManager = createCatalogManager(null);
+ ObjectIdentifier identifier =
+ ObjectIdentifier.of(
+ catalogManager.getCurrentCatalog(),
+ catalogManager.getCurrentDatabase(),
+ "connection");
+ catalogManager
+ .getCatalog(catalogManager.getCurrentCatalog())
+ .orElseThrow()
+ .createConnection(
+ identifier.toObjectPath(),
+ CatalogConnection.of(Map.of("endpoint", "permanent"),
null),
+ false);
+ catalogManager.createTemporaryConnection(
+ SensitiveConnection.of(Map.of("type", "default", "endpoint",
"temporary"), null),
+ identifier,
+ false);
+
+ ContextResolvedConnection temporaryConnection =
+ catalogManager.getResolvedConnection(identifier).orElseThrow();
+ assertThat(temporaryConnection.isTemporary()).isTrue();
+ assertThat(temporaryConnection.getConnection().getOptions())
+ .containsEntry("endpoint", "temporary");
+
+ catalogManager.dropTemporaryConnection(identifier, false);
+
+ ContextResolvedConnection permanentConnection =
+ catalogManager.getResolvedConnection(identifier).orElseThrow();
+ assertThat(permanentConnection.isTemporary()).isFalse();
+ assertThat(permanentConnection.getConnection().getOptions())
+ .containsEntry("endpoint", "permanent");
+ assertThat(
+ catalogManager.getResolvedConnection(
+ ObjectIdentifier.of(
+ catalogManager.getCurrentCatalog(),
+ catalogManager.getCurrentDatabase(),
+ "missing")))
+ .isEmpty();
+ }
+
@Test
public void testCreateConnectionWithoutTypeFallsBackToDefaultFactory()
throws Exception {
CompletableFuture<CreateConnectionEvent> createFuture = new
CompletableFuture<>();
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/DefaultConnectionFactory.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/DefaultConnectionFactory.java
index 236e8c6a5f4..c098bc6cc4f 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/DefaultConnectionFactory.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/DefaultConnectionFactory.java
@@ -64,7 +64,7 @@ public class DefaultConnectionFactory implements
ConnectionFactory {
* surrounding double underscores make collision with user-supplied option
names unlikely; user
* options containing this key will be rejected at create-time.
*/
- static final String SECRET_REFERENCE_KEY =
"__flink.encrypted-secret-key__";
+ public static final String SECRET_REFERENCE_KEY =
"__flink.encrypted-secret-key__";
/**
* Default whitelist of option keys treated as sensitive. Seeded from
{@link
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlNodeConverters.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlNodeConverters.java
index ce8f41df992..9faecf21871 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlNodeConverters.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlNodeConverters.java
@@ -141,6 +141,7 @@ public class SqlNodeConverters {
private static void registerConnectionConverters() {
register(new SqlCreateConnectionConverter());
+ register(new SqlShowCreateConnectionConverter());
}
private static void registerMaterializedTableConverters() {
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlShowCreateConnectionConverter.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlShowCreateConnectionConverter.java
new file mode 100644
index 00000000000..35fcaa32f3d
--- /dev/null
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlShowCreateConnectionConverter.java
@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.operations.converters;
+
+import org.apache.flink.sql.parser.dql.SqlShowCreateConnection;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.ContextResolvedConnection;
+import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.catalog.UnresolvedIdentifier;
+import org.apache.flink.table.operations.Operation;
+import org.apache.flink.table.operations.ShowCreateConnectionOperation;
+
+import java.util.Optional;
+
+/** A converter for {@link SqlShowCreateConnection}. */
+public class SqlShowCreateConnectionConverter implements
SqlNodeConverter<SqlShowCreateConnection> {
+
+ @Override
+ public Operation convertSqlNode(
+ SqlShowCreateConnection showCreateConnection, ConvertContext
context) {
+ UnresolvedIdentifier unresolvedIdentifier =
+
UnresolvedIdentifier.of(showCreateConnection.getConnectionName().names);
+ ObjectIdentifier identifier =
+
context.getCatalogManager().qualifyIdentifier(unresolvedIdentifier);
+ Optional<ContextResolvedConnection> connection =
+ context.getCatalogManager().getResolvedConnection(identifier);
+
+ if (connection.isEmpty()) {
+ throw new ValidationException(
+ String.format(
+ "Could not execute SHOW CREATE CONNECTION.
Connection with identifier %s does not exist.",
+ identifier.asSerializableString()));
+ }
+
+ ContextResolvedConnection resolvedConnection = connection.get();
+ return new ShowCreateConnectionOperation(
+ identifier, resolvedConnection.getConnection(),
resolvedConnection.isTemporary());
+ }
+}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlConnectionOperationConverterTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlConnectionOperationConverterTest.java
index a410c95ec0b..fbed59026ae 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlConnectionOperationConverterTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlConnectionOperationConverterTest.java
@@ -20,9 +20,11 @@ package org.apache.flink.table.planner.operations;
import org.apache.flink.sql.parser.error.SqlValidateException;
import org.apache.flink.table.api.SqlParserException;
+import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.SensitiveConnection;
import org.apache.flink.table.operations.Operation;
+import org.apache.flink.table.operations.ShowCreateConnectionOperation;
import org.apache.flink.table.operations.ddl.CreateConnectionOperation;
import org.junit.jupiter.api.Test;
@@ -121,4 +123,50 @@ class SqlConnectionOperationConverterTest extends
SqlNodeToOperationConversionTe
.isInstanceOf(SqlValidateException.class)
.hasMessageContaining("Connection property list can not be
empty.");
}
+
+ @Test
+ void testShowCreateConnection() {
+ createTemporaryConnection("my_conn");
+
+ Operation operation = parse("SHOW CREATE CONNECTION my_conn");
+
assertThat(operation).isInstanceOf(ShowCreateConnectionOperation.class);
+ ShowCreateConnectionOperation op = (ShowCreateConnectionOperation)
operation;
+
+ assertThat(op.getConnectionIdentifier())
+ .isEqualTo(ObjectIdentifier.of("builtin", "default",
"my_conn"));
+ assertThat(op.getConnection().getOptions()).containsEntry("k", "v");
+
assertThat(op.getConnection().getOptions()).doesNotContainKey("password");
+ assertThat(op.isTemporary()).isTrue();
+ }
+
+ @Test
+ void testShowCreateConnectionWithFullyQualifiedName() {
+ createTemporaryConnection("my_conn");
+
+ Operation operation = parse("SHOW CREATE CONNECTION
`builtin`.`default`.`my_conn`");
+ ShowCreateConnectionOperation op = (ShowCreateConnectionOperation)
operation;
+
+ assertThat(op.getConnectionIdentifier())
+ .isEqualTo(ObjectIdentifier.of("builtin", "default",
"my_conn"));
+ }
+
+ @Test
+ void testShowCreateConnectionMissingConnectionRejected() {
+ assertThatThrownBy(() -> parse("SHOW CREATE CONNECTION missing_conn"))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining(
+ "Could not execute SHOW CREATE CONNECTION. Connection
with identifier");
+ }
+
+ private void createTemporaryConnection(String connectionName) {
+ catalogManager.createTemporaryConnection(
+ SensitiveConnection.of(
+ Map.of("type", "default", "k", "v", "password",
"super-secret"),
+ "hi there"),
+ ObjectIdentifier.of(
+ catalogManager.getCurrentCatalog(),
+ catalogManager.getCurrentDatabase(),
+ connectionName),
+ false);
+ }
}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/CreateConnectionITCase.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/CreateConnectionITCase.java
index a9d99cbfe5d..170768b656d 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/CreateConnectionITCase.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/CreateConnectionITCase.java
@@ -18,14 +18,22 @@
package org.apache.flink.table.planner.runtime.batch.sql;
+import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.api.internal.TableEnvironmentInternal;
+import org.apache.flink.table.catalog.CatalogConnection;
import org.apache.flink.table.catalog.CatalogManager;
import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.factories.DefaultConnectionFactory;
import org.apache.flink.table.planner.runtime.utils.BatchTestBase;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CollectionUtil;
import org.junit.jupiter.api.Test;
+import java.util.List;
+import java.util.Map;
+
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.assertj.core.api.Assertions.entry;
@@ -73,6 +81,91 @@ class CreateConnectionITCase extends BatchTestBase {
.hasMessageContaining("WritableSecretStore must be
configured");
}
+ @Test
+ void testShowCreatePermanentConnection() throws Exception {
+ ObjectIdentifier identifier = connectionIdentifier("my_conn");
+ catalogManager()
+ .getCatalog(catalogManager().getCurrentCatalog())
+ .orElseThrow()
+ .createConnection(
+ identifier.toObjectPath(),
+ CatalogConnection.of(Map.of("k", "v", "type",
"default"), "hi there"),
+ false);
+
+ List<Row> rows = collectRows("SHOW CREATE CONNECTION my_conn");
+
+ assertThat(rows).hasSize(1);
+ String showCreate = (String) rows.get(0).getField(0);
+ assertThat(showCreate)
+ .contains("CREATE CONNECTION")
+ .doesNotContain("CREATE TEMPORARY CONNECTION")
+ .contains("`my_conn`")
+ .contains("COMMENT 'hi there'")
+ .contains("'k' = 'v'")
+ .contains("'type' = 'default'");
+ }
+
+ @Test
+ void testShowCreateTemporaryConnection() {
+ tEnv().executeSql(
+ "CREATE TEMPORARY CONNECTION my_conn COMMENT 'hi
there' "
+ + "WITH ('type' = 'default', 'k' = 'v',
'password' = 'super-secret')");
+
+
assertThat(catalogManager().getConnection(connectionIdentifier("my_conn")))
+ .hasValueSatisfying(
+ connection ->
+ assertThat(connection.getOptions())
+ .containsKeys(
+ "k",
+ "type",
+
DefaultConnectionFactory.SECRET_REFERENCE_KEY)
+ .doesNotContainKey("password"));
+
+ List<Row> rows = collectRows("SHOW CREATE CONNECTION my_conn");
+
+ assertThat(rows).hasSize(1);
+ String showCreate = (String) rows.get(0).getField(0);
+ assertThat(showCreate)
+ .contains("CREATE TEMPORARY CONNECTION")
+ .contains("`my_conn`")
+ .contains("COMMENT 'hi there'")
+ .contains("'k' = 'v'")
+ .contains("'type' = 'default'")
+ .doesNotContain("super-secret")
+ .doesNotContain("password")
+ .doesNotContain(DefaultConnectionFactory.SECRET_REFERENCE_KEY);
+ }
+
+ @Test
+ void testShowCreateSecretOnlyTemporaryConnection() {
+ tEnv().executeSql("CREATE TEMPORARY CONNECTION my_conn WITH
('password' = 'super-secret')");
+
+ List<Row> rows = collectRows("SHOW CREATE CONNECTION my_conn");
+
+ assertThat(rows).hasSize(1);
+ String showCreate = (String) rows.get(0).getField(0);
+ assertThat(showCreate)
+ .contains("CREATE TEMPORARY CONNECTION")
+ .contains("WITH (\n 'type' = 'default'\n)\n")
+ .doesNotContain("password")
+ .doesNotContain("super-secret")
+ .doesNotContain(DefaultConnectionFactory.SECRET_REFERENCE_KEY);
+
+
catalogManager().dropTemporaryConnection(connectionIdentifier("my_conn"),
false);
+ tEnv().executeSql(showCreate);
+
+
assertThat(catalogManager().getConnection(connectionIdentifier("my_conn")))
+ .hasValueSatisfying(
+ connection ->
+ assertThat(connection.getOptions())
+ .containsOnly(entry("type",
"default")));
+ }
+
+ private List<Row> collectRows(String sql) {
+ TableResult result = tEnv().executeSql(sql);
+ return CollectionUtil.iteratorToList(result.collect());
+ }
+
private CatalogManager catalogManager() {
return ((TableEnvironmentInternal) tEnv()).getCatalogManager();
}
