[FLINK-38262][table] show connections operation

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

@@ -0,0 +1,91 @@
+/*
+ * 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.table.catalog.CatalogManager;
+import org.apache.flink.table.operations.utils.ShowLikeOperator;
+
+import javax.annotation.Nullable;
+
+import java.util.Set;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Operation to describe a SHOW CONNECTIONS statement. The full syntax for SHOW CONNECTIONS is as
+ * follows:
+ *
+ * <pre>{@code
+ * SHOW CONNECTIONS [ ( FROM | IN ) [catalog_name.]database_name ] [[NOT] LIKE
+ * <sql_like_pattern>]
+ * }</pre>
+ */
+@Internal
+public class ShowConnectionsOperation extends AbstractShowOperation {
+
+    private final @Nullable String databaseName;
+
+    public ShowConnectionsOperation(
+            @Nullable String catalogName,
+            @Nullable String databaseName,
+            @Nullable String preposition,
+            @Nullable ShowLikeOperator likeOp) {
+        super(catalogName, preposition, likeOp);
+        this.databaseName = databaseName;
+    }
+
+    public ShowConnectionsOperation(
+            @Nullable String catalogName,
+            @Nullable String databaseName,
+            @Nullable ShowLikeOperator likeOp) {
+        this(catalogName, databaseName, null, likeOp);
+    }
+
+    @Nullable
+    public String getDatabaseName() {
+        return databaseName;
+    }
+
+    @Override
+    protected Set<String> retrieveDataForTableResult(Context ctx) {
+        final CatalogManager catalogManager = ctx.getCatalogManager();
+        return catalogManager.listConnections(
+                checkNotNull(catalogName, "catalogName"),
+                checkNotNull(databaseName, "databaseName"));
+    }
+
+    @Override
+    protected String getOperationName() {
+        return "SHOW CONNECTIONS";
+    }
+
+    @Override
+    protected String getColumnName() {
+        return "connection name";
+    }
+
+    @Override
+    public String getPrepositionSummaryString() {
+        if (databaseName == null) {
+            return super.getPrepositionSummaryString();
+        }
+        return super.getPrepositionSummaryString() + "." + databaseName;
+    }
+}

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

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlShowConnectionsConverter.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.planner.operations.converters;
+
+import org.apache.flink.sql.parser.dql.SqlShowConnections;
+import org.apache.flink.table.operations.Operation;
+import org.apache.flink.table.operations.ShowConnectionsOperation;
+import org.apache.flink.table.operations.utils.ShowLikeOperator;
+
+import javax.annotation.Nullable;
+
+/** A converter for {@link SqlShowConnections}. */
+public class SqlShowConnectionsConverter extends AbstractSqlShowConverter<SqlShowConnections> {
+
+    @Override
+    public Operation getOperationWithoutPrep(
+            SqlShowConnections sqlShowCall,
+            @Nullable String catalogName,
+            @Nullable String databaseName,
+            @Nullable ShowLikeOperator likeOp) {
+        return new ShowConnectionsOperation(catalogName, databaseName, likeOp);
+    }
+
+    @Override
+    public Operation getOperation(
+            SqlShowConnections sqlShowCall,
+            @Nullable String catalogName,
+            @Nullable String databaseName,
+            String prep,
+            @Nullable ShowLikeOperator likeOp) {
+        return new ShowConnectionsOperation(catalogName, databaseName, prep, likeOp);
+    }
+
+    @Override
+    public Operation convertSqlNode(SqlShowConnections sqlShowConnections, ConvertContext context) {
+        return convertShowOperation(sqlShowConnections, context);
+    }
+}

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

@@ -141,6 +141,7 @@ class FlinkPlannerImpl(
         || sqlNode.isInstanceOf[SqlShowCurrentDatabase]
         || sqlNode.isInstanceOf[SqlShowTables]
         || sqlNode.isInstanceOf[SqlShowModels]
+        || sqlNode.isInstanceOf[SqlShowConnections]
         || sqlNode.isInstanceOf[SqlShowFunctions]
         || sqlNode.isInstanceOf[SqlShowJars]
         || sqlNode.isInstanceOf[SqlShowModules]

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

@@ -23,7 +23,10 @@
 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.ShowConnectionsOperation;
 import org.apache.flink.table.operations.ddl.CreateConnectionOperation;
+import org.apache.flink.table.operations.utils.LikeType;
+import org.apache.flink.table.operations.utils.ShowLikeOperator;
 
 import org.junit.jupiter.api.Test;
 
@@ -121,4 +124,45 @@ void testCreateConnectionWithEmptyOptionsRejected() {
                 .isInstanceOf(SqlValidateException.class)
                 .hasMessageContaining("Connection property list can not be empty.");
     }
+
+    @Test
+    void testShowConnections() {
+        Operation operation = parse("SHOW CONNECTIONS");
+        assertThat(operation).isInstanceOf(ShowConnectionsOperation.class);
+        assertThat(operation.asSummaryString()).isEqualTo("SHOW CONNECTIONS");
+    }
+
+    @Test
+    void testShowConnectionsWithDatabase() {
+        Operation operation = parse("SHOW CONNECTIONS FROM db1");
+        assertThat(operation).isInstanceOf(ShowConnectionsOperation.class);
+        assertThat(operation.asSummaryString()).isEqualTo("SHOW CONNECTIONS FROM builtin.db1");
+    }
+
+    @Test
+    void testShowConnectionsWithCatalogAndDatabase() {
+        Operation operation = parse("SHOW CONNECTIONS IN cat1.db1");
+        assertThat(operation).isInstanceOf(ShowConnectionsOperation.class);
+        assertThat(operation.asSummaryString()).isEqualTo("SHOW CONNECTIONS IN cat1.db1");
+    }
+
+    @Test
+    void testShowConnectionsLike() {
+        Operation operation = parse("SHOW CONNECTIONS LIKE '%conn%'");
+        assertThat(operation).isInstanceOf(ShowConnectionsOperation.class);
+        assertThat(operation.asSummaryString()).isEqualTo("SHOW CONNECTIONS LIKE '%conn%'");
+    }
+
+    @Test
+    void testShowConnectionsNotLike() {
+        Operation operation = parse("SHOW CONNECTIONS NOT LIKE 'tmp_%'");
+        assertThat(operation).isInstanceOf(ShowConnectionsOperation.class);
+        assertThat(operation)
+                .isEqualTo(
+                        new ShowConnectionsOperation(
+                                "builtin",
+                                "default",
+                                ShowLikeOperator.of(LikeType.NOT_LIKE, "tmp_%")));
+        assertThat(operation.asSummaryString()).isEqualTo("SHOW CONNECTIONS NOT LIKE 'tmp_%'");
+    }
 }

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,10 +78,35 @@ void testCreatePermanentConnectionRejectedWithoutSecretStore() {
                 .hasMessageContaining("WritableSecretStore must be configured");
     }
 
+    @Test
+    void testShowConnections() {
+        tEnv().executeSql("CREATE TEMPORARY CONNECTION b_conn WITH ('k' = 'v')");
+        tEnv().executeSql("CREATE TEMPORARY CONNECTION a_conn WITH ('k' = 'v')");
+
+        assertThat(collectRows("SHOW CONNECTIONS"))
+                .containsExactly(Row.of("a_conn"), Row.of("b_conn"));
+    }
+
+    @Test
+    void testShowConnectionsLike() {
+        tEnv().executeSql("CREATE TEMPORARY CONNECTION prod_conn WITH ('k' = 'v')");
+        tEnv().executeSql("CREATE TEMPORARY CONNECTION tmp_conn WITH ('k' = 'v')");
+
+        assertThat(collectRows("SHOW CONNECTIONS LIKE 'prod_%'"))
+                .containsExactly(Row.of("prod_conn"));
+        assertThat(collectRows("SHOW CONNECTIONS NOT LIKE 'prod_%'"))
+                .containsExactly(Row.of("tmp_conn"));
+    }
+
     private CatalogManager catalogManager() {
         return ((TableEnvironmentInternal) tEnv()).getCatalogManager();
     }
 
+    private List<Row> collectRows(String sql) {
+        TableResult result = tEnv().executeSql(sql);
+        return CollectionUtil.iteratorToList(result.collect());
+    }
+
     private ObjectIdentifier connectionIdentifier(String connectionName) {
         CatalogManager catalogManager = catalogManager();
         return ObjectIdentifier.of(