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(