lihaosky commented on code in PR #28792:
URL: https://github.com/apache/flink/pull/28792#discussion_r4075207630
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/CreateConnectionITCase.java:
##########
@@ -161,6 +161,38 @@ void testShowCreateSecretOnlyTemporaryConnection() {
.containsOnly(entry("type",
"default")));
}
+ @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(
Review Comment:
nit: the rows are sorted via `TreeMap`, so the output is deterministic.
`containsExactly` would pin the full output and also catch leaked or duplicated
rows, which makes the `noneMatch` checks below mostly redundant.
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/CreateConnectionITCase.java:
##########
@@ -161,6 +161,38 @@ void testShowCreateSecretOnlyTemporaryConnection() {
.containsOnly(entry("type",
"default")));
}
+ @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"));
Review Comment:
nit: nothing asserts `temporary` = `false` for a permanent connection.
`CREATE CONNECTION` fails in this test base because no secret store is
configured, but the connection can be registered directly through the catalog
like `testShowCreatePermanentConnection` does.
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/DescribeConnectionOperation.java:
##########
@@ -0,0 +1,118 @@
+/*
+ * 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.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.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 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(),
+
ctx.getTableConfig().get(SecurityOptions.ADDITIONAL_SENSITIVE_KEYS)));
+ }
+
+ private Object[][] buildRows(
+ CatalogConnection connection,
+ boolean isTemporary,
+ List<String> additionalSensitiveKeys) {
+ List<Object[]> rows = new ArrayList<>();
+ new
TreeMap<>(ShowCreateUtil.withoutConnectionInternalOptions(connection.getOptions()))
Review Comment:
When every option is sensitive, e.g. `CREATE TEMPORARY CONNECTION c WITH
('password' = 'x')`, `withoutConnectionInternalOptions` leaves an empty map and
`DESCRIBE CONNECTION c` returns zero rows for a connection that exists. `SHOW
CREATE CONNECTION` handles the same case in
`ShowCreateUtil.buildShowCreateConnectionRow` by falling back to `'type' =
'default'`.
I think DESCRIBE should always emit a `type` row (from
`FactoryUtil.CONNECTION_TYPE`, defaulting to `default`), similar to how
`DescribeCatalogOperation` always emits `type`. That also keeps DESCRIBE and
SHOW CREATE consistent for secret-only connections.
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/DescribeConnectionOperation.java:
##########
@@ -0,0 +1,118 @@
+/*
+ * 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.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.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 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(),
+
ctx.getTableConfig().get(SecurityOptions.ADDITIONAL_SENSITIVE_KEYS)));
+ }
+
+ private Object[][] buildRows(
+ CatalogConnection connection,
+ boolean isTemporary,
+ List<String> additionalSensitiveKeys) {
+ List<Object[]> rows = new ArrayList<>();
+ new
TreeMap<>(ShowCreateUtil.withoutConnectionInternalOptions(connection.getOptions()))
+ .forEach(
+ (key, value) -> {
+ rows.add(
+ new Object[] {
+ key, maskValue(key, value,
additionalSensitiveKeys)
+ });
+ });
+ if (connection.getComment() != null &&
!connection.getComment().isEmpty()) {
Review Comment:
nit: `StringUtils.isNotEmpty(connection.getComment())` would be simpler
here, and matches what `ShowCreateUtil.extractComment` does for SHOW CREATE.
##########
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
Review Comment:
nit: the sibling converters for `SqlRichDescribeModel` and
`SqlRichDescribeFunction` are named `SqlDescribeModelConverter` and
`SqlDescribeFunctionConverter`. Consider `SqlDescribeConnectionConverter` for
consistency.
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/CreateConnectionITCase.java:
##########
@@ -161,6 +161,38 @@ void testShowCreateSecretOnlyTemporaryConnection() {
.containsOnly(entry("type",
"default")));
}
+ @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"))
Review Comment:
This passes because `DefaultConnectionFactory.createConnection` already
removed the `password` key at CREATE time, not because of `maskValue`. As far
as I can tell no test exercises the masking branch at all.
If masking stays (see my comment on `maskValue` in
`DescribeConnectionOperation`), please add a case that actually hits it. For
example `'db.password' = 'x'`: the factory whitelist is exact-match so that key
stays in the catalog options, while `GlobalConfiguration.isSensitive` matches
by substring and redacts it. The test should assert the value is `******`, not
just that the key is absent.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
