diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ShowConnectionsOperation.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ShowConnectionsOperation.java
new file mode 100644
index 00000000000000..efaa86cdf76f4c
--- /dev/null
+++ b/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:
+ *
+ *
{@code
+ * SHOW CONNECTIONS [ ( FROM | IN ) [catalog_name.]database_name ] [[NOT] LIKE
+ * ]
+ * }
+ */
+@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 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;
+ }
+}
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..2d914098823c64 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 SqlShowConnectionsConverter());
}
private static void registerMaterializedTableConverters() {
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlShowConnectionsConverter.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlShowConnectionsConverter.java
new file mode 100644
index 00000000000000..789314e0cb7f67
--- /dev/null
+++ b/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 {
+
+ @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);
+ }
+}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/calcite/FlinkPlannerImpl.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/calcite/FlinkPlannerImpl.scala
index ee329a0ecd6f80..3569c36fa7a3e6 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/calcite/FlinkPlannerImpl.scala
+++ b/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]
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..02b9cc94da7149 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
@@ -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_%'");
+ }
}
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..e8424e55d52997 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,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 collectRows(String sql) {
+ TableResult result = tEnv().executeSql(sql);
+ return CollectionUtil.iteratorToList(result.collect());
+ }
+
private ObjectIdentifier connectionIdentifier(String connectionName) {
CatalogManager catalogManager = catalogManager();
return ObjectIdentifier.of(