This is an automated email from the ASF dual-hosted git repository.
lihaosky pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 4b568e71e1c [FLINK-38262][table] describe connection operation (#28792)
4b568e71e1c is described below
commit 4b568e71e1c6e79a43cdd3739207b1efd5ea351c
Author: Shekhar Rajak <[email protected]>
AuthorDate: Sat Oct 3 04:43:36 2026 +0530
[FLINK-38262][table] describe connection operation (#28792)
* [FLINK-38262][table] Add DESCRIBE CONNECTION operation
* [FLINK-38262][table] Test DESCRIBE CONNECTION conversion
* [FLINK-38262][table] Include default type in connection description
* [FLINK-38262][table] Disambiguate connection options in DESCRIBE
* [FLINK-38262][table] Always report connection scope in DESCRIBE
* [FLINK-38262][table] Simplify connection comment handling
* [FLINK-38262][table] Align connection converter naming
* [FLINK-38262][table] Assert complete connection description
* [FLINK-38262][table] Rely on factory-managed connection secrets
* [FLINK-38262][table] Test permanent connection description
---
.../flink/table/api/internal/ShowCreateUtil.java | 2 +-
.../operations/DescribeConnectionOperation.java | 108 +++++++++++++++++++++
.../converters/SqlDescribeConnectionConverter.java | 39 ++++++++
.../operations/converters/SqlNodeConverters.java | 1 +
.../table/planner/calcite/FlinkPlannerImpl.scala | 1 +
.../SqlConnectionOperationConverterTest.java | 33 +++++++
.../runtime/batch/sql/CreateConnectionITCase.java | 51 ++++++++++
7 files changed, 234 insertions(+), 1 deletion(-)
diff --git
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/ShowCreateUtil.java
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/ShowCreateUtil.java
index d8c0fa9306d..3b046eab13f 100644
---
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/ShowCreateUtil.java
+++
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/ShowCreateUtil.java
@@ -436,7 +436,7 @@ public class ShowCreateUtil {
return StringUtils.isEmpty(comment) ? Optional.empty() :
Optional.of(comment);
}
- private static Map<String, String> withoutConnectionInternalOptions(
+ public static Map<String, String> withoutConnectionInternalOptions(
Map<String, String> options) {
return options.entrySet().stream()
.filter(
diff --git
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/DescribeConnectionOperation.java
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/DescribeConnectionOperation.java
new file mode 100644
index 00000000000..ce1855c675c
--- /dev/null
+++
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/DescribeConnectionOperation.java
@@ -0,0 +1,108 @@
+/*
+ * 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.api.DataTypes;
+import org.apache.flink.table.api.ValidationException;
+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.ContextResolvedConnection;
+import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.types.DataType;
+
+import org.apache.commons.lang3.StringUtils;
+
+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 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) {
+ ContextResolvedConnection resolvedConnection =
+ ctx.getCatalogManager()
+ .getResolvedConnection(connectionIdentifier)
+ .orElseThrow(
+ () ->
+ new ValidationException(
+ String.format(
+ "Connection with
identifier '%s' does not exist.",
+
connectionIdentifier.asSummaryString())));
+ return buildTableResult(
+ new String[] {"name", "value"},
+ new DataType[] {DataTypes.STRING(), DataTypes.STRING()},
+ buildRows(resolvedConnection.getConnection(),
resolvedConnection.isTemporary()));
+ }
+
+ private Object[][] buildRows(CatalogConnection connection, boolean
isTemporary) {
+ List<Object[]> rows = new ArrayList<>();
+ TreeMap<String, String> options =
+ new TreeMap<>(
+
ShowCreateUtil.withoutConnectionInternalOptions(connection.getOptions()));
+ options.putIfAbsent(
+ FactoryUtil.CONNECTION_TYPE.key(),
FactoryUtil.CONNECTION_TYPE.defaultValue());
+ rows.add(
+ new Object[] {
+ FactoryUtil.CONNECTION_TYPE.key(),
+ options.remove(FactoryUtil.CONNECTION_TYPE.key())
+ });
+ options.forEach((key, value) -> rows.add(new Object[] {"option:" +
key, value}));
+ if (StringUtils.isNotEmpty(connection.getComment())) {
+ rows.add(new Object[] {"comment", connection.getComment()});
+ }
+ rows.add(new Object[] {"temporary", String.valueOf(isTemporary)});
+ return rows.toArray(new Object[0][]);
+ }
+}
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlDescribeConnectionConverter.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlDescribeConnectionConverter.java
new file mode 100644
index 00000000000..346c6cccf81
--- /dev/null
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlDescribeConnectionConverter.java
@@ -0,0 +1,39 @@
+/*
+ * 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 SqlDescribeConnectionConverter 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());
+ }
+}
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 9faecf21871..6ba372e5184 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 @@ public class SqlNodeConverters {
private static void registerConnectionConverters() {
register(new SqlCreateConnectionConverter());
+ register(new SqlDescribeConnectionConverter());
register(new SqlShowCreateConnectionConverter());
}
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 ee329a0ecd6..846402a940e 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
@@ -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]
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 fbed59026ae..42eb641dccf 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,13 +23,18 @@ 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.DescribeConnectionOperation;
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;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
import java.util.Map;
+import java.util.stream.Stream;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -158,6 +163,34 @@ class SqlConnectionOperationConverterTest extends
SqlNodeToOperationConversionTe
"Could not execute SHOW CREATE CONNECTION. Connection
with identifier");
}
+ @ParameterizedTest
+ @MethodSource("describeConnectionCases")
+ void testDescribeConnection(
+ String statement, ObjectIdentifier expectedIdentifier, boolean
expectedExtended) {
+ Operation operation = parse(statement);
+ assertThat(operation).isInstanceOf(DescribeConnectionOperation.class);
+ DescribeConnectionOperation op = (DescribeConnectionOperation)
operation;
+
+ assertThat(op.getConnectionIdentifier()).isEqualTo(expectedIdentifier);
+ assertThat(op.isExtended()).isEqualTo(expectedExtended);
+ }
+
+ private static Stream<Arguments> describeConnectionCases() {
+ return Stream.of(
+ Arguments.of(
+ "DESCRIBE CONNECTION my_conn",
+ ObjectIdentifier.of("builtin", "default", "my_conn"),
+ false),
+ Arguments.of(
+ "DESC CONNECTION EXTENDED my_conn",
+ ObjectIdentifier.of("builtin", "default", "my_conn"),
+ true),
+ Arguments.of(
+ "DESCRIBE CONNECTION cat1.db1.my_conn",
+ ObjectIdentifier.of("cat1", "db1", "my_conn"),
+ false));
+ }
+
private void createTemporaryConnection(String connectionName) {
catalogManager.createTemporaryConnection(
SensitiveConnection.of(
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 170768b656d..31150551da0 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
@@ -161,6 +161,57 @@ class CreateConnectionITCase extends BatchTestBase {
.containsOnly(entry("type",
"default")));
}
+ @Test
+ void testDescribeTemporaryConnection() {
+ tEnv().executeSql(
+ "CREATE TEMPORARY CONNECTION my_conn COMMENT 'hi
there' "
+ + "WITH ('type' = 'default', 'k' = 'v',
'comment' = 'option comment', "
+ + "'password' = 'super-secret')");
+
+ List<Row> rows = collectRows("DESCRIBE CONNECTION my_conn");
+
+ assertThat(rows)
+ .containsExactly(
+ Row.of("type", "default"),
+ Row.of("option:comment", "option comment"),
+ Row.of("option:k", "v"),
+ Row.of("comment", "hi there"),
+ Row.of("temporary", "true"));
+ }
+
+ @Test
+ void testDescribeSecretOnlyConnectionIncludesDefaultType() {
+ tEnv().executeSql("CREATE TEMPORARY CONNECTION my_conn WITH
('password' = 'secret')");
+
+ assertThat(collectRows("DESCRIBE CONNECTION my_conn"))
+ .containsExactly(Row.of("type", "default"),
Row.of("temporary", "true"));
+ }
+
+ @Test
+ void testDescribePermanentConnectionIncludesScope() throws Exception {
+ ObjectIdentifier identifier = connectionIdentifier("my_conn");
+ catalogManager()
+ .getCatalog(catalogManager().getCurrentCatalog())
+ .orElseThrow()
+ .createConnection(
+ identifier.toObjectPath(),
+ CatalogConnection.of(Map.of("k", "v"), null),
+ false);
+
+ assertThat(collectRows("DESCRIBE CONNECTION my_conn"))
+ .containsExactly(
+ Row.of("type", "default"),
+ Row.of("option:k", "v"),
+ Row.of("temporary", "false"));
+ }
+
+ @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());
