From 0b21c6efe97ac3a2c04782a1457cacf24335d1db Mon Sep 17 00:00:00 2001 From: shekharrajak Date: Tue, 21 Jul 2026 22:00:14 +0530 Subject: [PATCH] [FLINK-38262][table] Add SHOW CREATE CONNECTION operation Generated-by: Codex (GPT-5) --- .../table/api/internal/ShowCreateUtil.java | 37 ++++++++++ .../flink/table/catalog/CatalogManager.java | 30 ++++++++ .../catalog/ContextResolvedConnection.java | 54 ++++++++++++++ .../ShowCreateConnectionOperation.java | 74 +++++++++++++++++++ .../converters/SqlNodeConverters.java | 1 + .../SqlShowCreateConnectionConverter.java | 55 ++++++++++++++ .../SqlConnectionOperationConverterTest.java | 48 ++++++++++++ .../batch/sql/CreateConnectionITCase.java | 31 ++++++++ 8 files changed, 330 insertions(+) create mode 100644 flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/ContextResolvedConnection.java create mode 100644 flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ShowCreateConnectionOperation.java create mode 100644 flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlShowCreateConnectionConverter.java 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 9ea15b1eca867f..7dad200ff4faac 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.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; @@ -64,6 +65,7 @@ public class ShowCreateUtil { private static final DateTimeFormatter TIMESTAMP_FORMATTER = DateTimeFormatter.ofPattern("uuuu-MM-dd HH:mm:ss"); private static final String PRINT_INDENT = " "; + private static final String CONNECTION_SECRET_REFERENCE_KEY = "__flink.encrypted-secret-key__"; private ShowCreateUtil() {} @@ -98,6 +100,29 @@ public static String buildShowCreateModelRow( return sb.toString(); } + public static String buildShowCreateConnectionRow( + CatalogConnection connection, + ObjectIdentifier connectionIdentifier, + boolean isTemporary, + List additionalSensitiveKeys) { + StringBuilder sb = + new StringBuilder() + .append( + buildCreateFormattedPrefix( + "CONNECTION", + isTemporary, + connectionIdentifier, + false, + false)); + extractComment(connection).ifPresent(c -> sb.append(formatComment(c)).append("\n")); + extractFormattedOptions( + withoutConnectionInternalOptions(connection.getOptions()), + 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, @@ -349,6 +374,18 @@ static Optional extractComment(ResolvedCatalogModel model) { : Optional.of(model.getComment()); } + static Optional extractComment(CatalogConnection connection) { + return StringUtils.isEmpty(connection.getComment()) + ? Optional.empty() + : Optional.of(connection.getComment()); + } + + static Map withoutConnectionInternalOptions(Map options) { + return options.entrySet().stream() + .filter(entry -> !CONNECTION_SECRET_REFERENCE_KEY.equals(entry.getKey())) + .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); + } + static Optional extractFormattedDistributedInfo(ResolvedCatalogTable catalogTable) { return catalogTable.getDistribution().map(TableDistribution::toString); } 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 ca31e17ebf4e48..cc9aafc90e32b8 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 @@ -1910,6 +1910,36 @@ public Optional getConnection(ObjectIdentifier objectIdentifi } } + /** + * Get a connection from the catalog with contextual metadata. + * + * @param objectIdentifier The fully qualified path of the connection. + * @return The requested connection wrapped in Optional. + */ + public Optional getResolvedConnection( + ObjectIdentifier objectIdentifier) { + CatalogConnection temporaryConnection = temporaryConnections.get(objectIdentifier); + if (temporaryConnection != null) { + return Optional.of( + ContextResolvedConnection.of(objectIdentifier, temporaryConnection, true)); + } + + Optional catalog = getCatalog(objectIdentifier.getCatalogName()); + if (catalog.isPresent()) { + try { + return Optional.of( + ContextResolvedConnection.of( + objectIdentifier, + catalog.get().getConnection(objectIdentifier.toObjectPath()), + false)); + } catch (ConnectionNotExistException | UnsupportedOperationException e) { + return Optional.empty(); + } + } else { + return Optional.empty(); + } + } + /** * 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 00000000000000..5df73f53d3584a --- /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 00000000000000..f3f25cd0b46ef9 --- /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-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 ce8f41df992295..9faecf21871759 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 @@ private static void registerCatalogConverters() { 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 00000000000000..35fcaa32f3d6d1 --- /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 { + + @Override + public Operation convertSqlNode( + SqlShowCreateConnection showCreateConnection, ConvertContext context) { + UnresolvedIdentifier unresolvedIdentifier = + UnresolvedIdentifier.of(showCreateConnection.getConnectionName().names); + ObjectIdentifier identifier = + context.getCatalogManager().qualifyIdentifier(unresolvedIdentifier); + Optional 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 a410c95ec0b659..fbed59026ae946 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 @@ 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 @@ void testCreateConnectionWithEmptyOptionsRejected() { .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 a9d99cbfe5d328..6954f249ed17ba 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,19 @@ 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.CatalogManager; import org.apache.flink.table.catalog.ObjectIdentifier; 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 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 +78,32 @@ void testCreatePermanentConnectionRejectedWithoutSecretStore() { .hasMessageContaining("WritableSecretStore must be configured"); } + @Test + void testShowCreateTemporaryConnection() { + tEnv().executeSql( + "CREATE TEMPORARY CONNECTION my_conn COMMENT 'hi there' " + + "WITH ('type' = 'default', 'k' = 'v', 'password' = 'super-secret')"); + + List 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("__flink.encrypted-secret-key__"); + } + + private List collectRows(String sql) { + TableResult result = tEnv().executeSql(sql); + return CollectionUtil.iteratorToList(result.collect()); + } + private CatalogManager catalogManager() { return ((TableEnvironmentInternal) tEnv()).getCatalogManager(); }