[FLINK-38262][table] describe connection operation

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

@@ -1910,6 +1910,10 @@ public Optional<CatalogConnection> getConnection(ObjectIdentifier objectIdentifi
         }
     }
 
+    public boolean isTemporaryConnection(ObjectIdentifier objectIdentifier) {
+        return temporaryConnections.containsKey(objectIdentifier);
+    }
+
     /**
      * List all connections in the given catalog and database.
      *

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/DescribeConnectionOperation.java

@@ -0,0 +1,122 @@
+/*
+ * 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.GlobalConfiguration;
+import org.apache.flink.configuration.SecurityOptions;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.api.internal.TableResultInternal;
+import org.apache.flink.table.catalog.CatalogConnection;
+import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.types.DataType;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
+
+import static org.apache.flink.table.api.internal.TableResultUtils.buildTableResult;
+
+/** Operation to describe a DESCRIBE CONNECTION statement. */
+@Internal
+public class DescribeConnectionOperation implements Operation, ExecutableOperation {
+
+    private static final String CONNECTION_SECRET_REFERENCE_KEY = "__flink.encrypted-secret-key__";
+
+    private final ObjectIdentifier connectionIdentifier;
+    private final boolean isExtended;
+
+    public DescribeConnectionOperation(ObjectIdentifier connectionIdentifier, boolean isExtended) {
+        this.connectionIdentifier = connectionIdentifier;
+        this.isExtended = isExtended;
+    }
+
+    public ObjectIdentifier getConnectionIdentifier() {
+        return connectionIdentifier;
+    }
+
+    public boolean isExtended() {
+        return isExtended;
+    }
+
+    @Override
+    public String asSummaryString() {
+        Map<String, Object> params = new LinkedHashMap<>();
+        params.put("identifier", connectionIdentifier);
+        params.put("isExtended", isExtended);
+        return OperationUtils.formatWithChildren(
+                "DESCRIBE CONNECTION", params, Collections.emptyList(), Operation::asSummaryString);
+    }
+
+    @Override
+    public TableResultInternal execute(Context ctx) {
+        CatalogConnection connection =
+                ctx.getCatalogManager()
+                        .getConnection(connectionIdentifier)
+                        .orElseThrow(
+                                () ->
+                                        new ValidationException(
+                                                String.format(
+                                                        "Connection with identifier '%s' does not exist.",
+                                                        connectionIdentifier.asSummaryString())));
+        boolean isTemporary = ctx.getCatalogManager().isTemporaryConnection(connectionIdentifier);
+
+        return buildTableResult(
+                new String[] {"name", "value"},
+                new DataType[] {DataTypes.STRING(), DataTypes.STRING()},
+                buildRows(
+                        connection,
+                        isTemporary,
+                        ctx.getTableConfig().get(SecurityOptions.ADDITIONAL_SENSITIVE_KEYS)));
+    }
+
+    private Object[][] buildRows(
+            CatalogConnection connection,
+            boolean isTemporary,
+            List<String> additionalSensitiveKeys) {
+        List<Object[]> rows = new ArrayList<>();
+        new TreeMap<>(connection.getOptions())
+                .forEach(
+                        (key, value) -> {
+                            if (!CONNECTION_SECRET_REFERENCE_KEY.equals(key)) {
+                                rows.add(
+                                        new Object[] {
+                                            key, maskValue(key, value, additionalSensitiveKeys)
+                                        });
+                            }
+                        });
+        if (connection.getComment() != null && !connection.getComment().isEmpty()) {
+            rows.add(new Object[] {"comment", connection.getComment()});
+        }
+        if (isExtended) {
+            rows.add(new Object[] {"temporary", String.valueOf(isTemporary)});
+        }
+        return rows.toArray(new Object[0][]);
+    }
+
+    private String maskValue(String key, String value, List<String> additionalSensitiveKeys) {
+        return GlobalConfiguration.isSensitive(key, additionalSensitiveKeys)
+                ? GlobalConfiguration.HIDDEN_CONTENT
+                : value;
+    }
+}

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 SqlRichDescribeConnectionConverter());
     }
 
     private static void registerMaterializedTableConverters() {

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlRichDescribeConnectionConverter.java

@@ -0,0 +1,40 @@
+/*
+ * 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.SqlRichDescribeConnection;
+import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.catalog.UnresolvedIdentifier;
+import org.apache.flink.table.operations.DescribeConnectionOperation;
+import org.apache.flink.table.operations.Operation;
+
+/** A converter for {@link SqlRichDescribeConnection}. */
+public class SqlRichDescribeConnectionConverter
+        implements SqlNodeConverter<SqlRichDescribeConnection> {
+
+    @Override
+    public Operation convertSqlNode(
+            SqlRichDescribeConnection sqlRichDescribeConnection, ConvertContext context) {
+        UnresolvedIdentifier unresolvedIdentifier =
+                UnresolvedIdentifier.of(sqlRichDescribeConnection.fullConnectionName());
+        ObjectIdentifier identifier =
+                context.getCatalogManager().qualifyIdentifier(unresolvedIdentifier);
+        return new DescribeConnectionOperation(identifier, sqlRichDescribeConnection.isExtended());
+    }
+}

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/calcite/FlinkPlannerImpl.scala

@@ -150,6 +150,7 @@ class FlinkPlannerImpl(
         || sqlNode.isInstanceOf[SqlShowJobs]
         || sqlNode.isInstanceOf[SqlDescribeJob]
         || sqlNode.isInstanceOf[SqlRichDescribeFunction]
+        || sqlNode.isInstanceOf[SqlRichDescribeConnection]
         || sqlNode.isInstanceOf[SqlRichDescribeModel]
         || sqlNode.isInstanceOf[SqlRichDescribeTable]
         || sqlNode.isInstanceOf[SqlUnloadModule]

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlConnectionOperationConverterTest.java

@@ -22,6 +22,7 @@
 import org.apache.flink.table.api.SqlParserException;
 import org.apache.flink.table.catalog.ObjectIdentifier;
 import org.apache.flink.table.catalog.SensitiveConnection;
+import org.apache.flink.table.operations.DescribeConnectionOperation;
 import org.apache.flink.table.operations.Operation;
 import org.apache.flink.table.operations.ddl.CreateConnectionOperation;
 
@@ -121,4 +122,30 @@ void testCreateConnectionWithEmptyOptionsRejected() {
                 .isInstanceOf(SqlValidateException.class)
                 .hasMessageContaining("Connection property list can not be empty.");
     }
+
+    @Test
+    void testDescribeConnection() {
+        Operation operation = parse("DESCRIBE CONNECTION my_conn");
+        assertThat(operation).isInstanceOf(DescribeConnectionOperation.class);
+        DescribeConnectionOperation op = (DescribeConnectionOperation) operation;
+
+        assertThat(op.getConnectionIdentifier())
+                .isEqualTo(ObjectIdentifier.of("builtin", "default", "my_conn"));
+        assertThat(op.isExtended()).isFalse();
+    }
+
+    @Test
+    void testDescribeConnectionExtended() {
+        Operation operation = parse("DESC CONNECTION EXTENDED my_conn");
+        DescribeConnectionOperation op = (DescribeConnectionOperation) operation;
+        assertThat(op.isExtended()).isTrue();
+    }
+
+    @Test
+    void testDescribeConnectionWithFullyQualifiedName() {
+        Operation operation = parse("DESCRIBE CONNECTION cat1.db1.my_conn");
+        DescribeConnectionOperation op = (DescribeConnectionOperation) operation;
+        assertThat(op.getConnectionIdentifier())
+                .isEqualTo(ObjectIdentifier.of("cat1", "db1", "my_conn"));
+    }
 }

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,43 @@ void testCreatePermanentConnectionRejectedWithoutSecretStore() {
                 .hasMessageContaining("WritableSecretStore must be configured");
     }
 
+    @Test
+    void testDescribeTemporaryConnection() {
+        tEnv().executeSql(
+                        "CREATE TEMPORARY CONNECTION my_conn COMMENT 'hi there' "
+                                + "WITH ('type' = 'default', 'k' = 'v', 'password' = 'super-secret')");
+
+        List<Row> rows = collectRows("DESCRIBE CONNECTION my_conn");
+
+        assertThat(rows)
+                .contains(
+                        Row.of("k", "v"), Row.of("type", "default"), Row.of("comment", "hi there"));
+        assertThat(rows.stream().map(Row::toString))
+                .noneMatch(row -> row.contains("super-secret"))
+                .noneMatch(row -> row.contains("password"))
+                .noneMatch(row -> row.contains("__flink.encrypted-secret-key__"));
+    }
+
+    @Test
+    void testDescribeTemporaryConnectionExtended() {
+        tEnv().executeSql("CREATE TEMPORARY CONNECTION my_conn WITH ('k' = 'v')");
+
+        assertThat(collectRows("DESCRIBE CONNECTION EXTENDED my_conn"))
+                .contains(Row.of("temporary", "true"));
+    }
+
+    @Test
+    void testDescribeMissingConnectionRejected() {
+        assertThatThrownBy(() -> tEnv().executeSql("DESCRIBE CONNECTION missing_conn"))
+                .isInstanceOf(ValidationException.class)
+                .hasMessageContaining("Connection with identifier");
+    }
+
+    private List<Row> collectRows(String sql) {
+        TableResult result = tEnv().executeSql(sql);
+        return CollectionUtil.iteratorToList(result.collect());
+    }
+
     private CatalogManager catalogManager() {
         return ((TableEnvironmentInternal) tEnv()).getCatalogManager();
     }