[FLINK-38262][table] show create connection operation

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<String> 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<String> extractComment(ResolvedCatalogModel model) {
                 : Optional.of(model.getComment());
     }
 
+    static Optional<String> extractComment(CatalogConnection connection) {
+        return StringUtils.isEmpty(connection.getComment())
+                ? Optional.empty()
+                : Optional.of(connection.getComment());
+    }
+
+    static Map<String, String> withoutConnectionInternalOptions(Map<String, String> 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<String> extractFormattedDistributedInfo(ResolvedCatalogTable catalogTable) {
         return catalogTable.getDistribution().map(TableDistribution::toString);
     }

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java

@@ -1910,6 +1910,36 @@ public Optional<CatalogConnection> 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<ContextResolvedConnection> getResolvedConnection(
+            ObjectIdentifier objectIdentifier) {
+        CatalogConnection temporaryConnection = temporaryConnections.get(objectIdentifier);
+        if (temporaryConnection != null) {
+            return Optional.of(
+                    ContextResolvedConnection.of(objectIdentifier, temporaryConnection, true));
+        }
+
+        Optional<Catalog> 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.
      *

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;
+    }
+}

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});
+    }
+}

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() {

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());
+    }
+}

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);
+    }
 }