This is an automated email from the ASF dual-hosted git repository.
snuyanzin 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 67a3291aeac [FLINK-38262][table] Add `CreateConnectionOperation` and
converter
67a3291aeac is described below
commit 67a3291aeac7d3075618294e12cd3bfe3b732261
Author: Shekhar Prasad Rajak <[email protected]>
AuthorDate: Tue Jul 21 17:01:28 2026 +0530
[FLINK-38262][table] Add `CreateConnectionOperation` and converter
---
.../operations/ddl/CreateConnectionOperation.java | 104 +++++++++++++++++
flink-table/flink-table-planner/pom.xml | 6 +
.../converters/SqlCreateConnectionConverter.java | 52 +++++++++
.../operations/converters/SqlNodeConverters.java | 5 +
.../SqlConnectionOperationConverterTest.java | 124 +++++++++++++++++++++
.../runtime/batch/sql/CreateConnectionITCase.java | 87 +++++++++++++++
6 files changed, 378 insertions(+)
diff --git
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/CreateConnectionOperation.java
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/CreateConnectionOperation.java
new file mode 100644
index 00000000000..b853150b955
--- /dev/null
+++
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/CreateConnectionOperation.java
@@ -0,0 +1,104 @@
+/*
+ * 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.ddl;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.internal.TableResultImpl;
+import org.apache.flink.table.api.internal.TableResultInternal;
+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.OperationUtils;
+
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/** Operation to describe a CREATE CONNECTION statement. */
+@Internal
+public class CreateConnectionOperation implements CreateOperation {
+
+ private static final String MASKED_VALUE = "****";
+
+ private final ObjectIdentifier connectionIdentifier;
+ private final SensitiveConnection sensitiveConnection;
+ private final boolean ignoreIfExists;
+ private final boolean isTemporary;
+
+ public CreateConnectionOperation(
+ ObjectIdentifier connectionIdentifier,
+ SensitiveConnection sensitiveConnection,
+ boolean ignoreIfExists,
+ boolean isTemporary) {
+ this.connectionIdentifier = connectionIdentifier;
+ this.sensitiveConnection = sensitiveConnection;
+ this.ignoreIfExists = ignoreIfExists;
+ this.isTemporary = isTemporary;
+ }
+
+ public ObjectIdentifier getConnectionIdentifier() {
+ return connectionIdentifier;
+ }
+
+ public SensitiveConnection getSensitiveConnection() {
+ return sensitiveConnection;
+ }
+
+ public boolean isIgnoreIfExists() {
+ return ignoreIfExists;
+ }
+
+ public boolean isTemporary() {
+ return isTemporary;
+ }
+
+ @Override
+ public String asSummaryString() {
+ Map<String, String> maskedOptions =
+ sensitiveConnection.getOptions().entrySet().stream()
+ .collect(
+ Collectors.toMap(
+ Map.Entry::getKey,
+ e -> MASKED_VALUE,
+ (a, b) -> a,
+ LinkedHashMap::new));
+ Map<String, Object> params = new LinkedHashMap<>();
+ params.put("connectionOptions", maskedOptions);
+ params.put("identifier", connectionIdentifier);
+ params.put("ignoreIfExists", ignoreIfExists);
+ params.put("isTemporary", isTemporary);
+
+ return OperationUtils.formatWithChildren(
+ "CREATE CONNECTION", params, List.of(),
Operation::asSummaryString);
+ }
+
+ @Override
+ public TableResultInternal execute(Context ctx) {
+ if (isTemporary) {
+ ctx.getCatalogManager()
+ .createTemporaryConnection(
+ sensitiveConnection, connectionIdentifier,
ignoreIfExists);
+ } else {
+ ctx.getCatalogManager()
+ .createConnection(sensitiveConnection,
connectionIdentifier, ignoreIfExists);
+ }
+ return TableResultImpl.TABLE_RESULT_OK;
+ }
+}
diff --git a/flink-table/flink-table-planner/pom.xml
b/flink-table/flink-table-planner/pom.xml
index b8f76f7f969..450dc204ec6 100644
--- a/flink-table/flink-table-planner/pom.xml
+++ b/flink-table/flink-table-planner/pom.xml
@@ -158,6 +158,12 @@ under the License.
<artifactId>flink-table-runtime</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-type-utils</artifactId>
+ <version>${project.version}</version>
+ <optional>${flink.markBundledAsOptional}</optional>
+ </dependency>
<!-- Table Calcite Bridge (included in the uber) -->
<dependency>
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlCreateConnectionConverter.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlCreateConnectionConverter.java
new file mode 100644
index 00000000000..f1bb4efa9ea
--- /dev/null
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlCreateConnectionConverter.java
@@ -0,0 +1,52 @@
+/*
+ * 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.ddl.connection.SqlCreateConnection;
+import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.catalog.SensitiveConnection;
+import org.apache.flink.table.catalog.UnresolvedIdentifier;
+import org.apache.flink.table.operations.Operation;
+import org.apache.flink.table.operations.ddl.CreateConnectionOperation;
+
+import java.util.Map;
+
+/** A converter for {@link SqlCreateConnection}. */
+public class SqlCreateConnectionConverter implements
SqlNodeConverter<SqlCreateConnection> {
+
+ @Override
+ public Operation convertSqlNode(
+ SqlCreateConnection sqlCreateConnection, ConvertContext context) {
+ UnresolvedIdentifier unresolvedIdentifier =
+ UnresolvedIdentifier.of(sqlCreateConnection.getFullName());
+ ObjectIdentifier identifier =
+
context.getCatalogManager().qualifyIdentifier(unresolvedIdentifier);
+
+ Map<String, String> options = sqlCreateConnection.getProperties();
+ String comment = sqlCreateConnection.getComment();
+
+ SensitiveConnection sensitiveConnection =
SensitiveConnection.of(options, comment);
+
+ return new CreateConnectionOperation(
+ identifier,
+ sensitiveConnection,
+ sqlCreateConnection.isIfNotExists(),
+ sqlCreateConnection.isTemporary());
+ }
+}
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 7e03c203664..ce8f41df992 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
@@ -90,6 +90,7 @@ public class SqlNodeConverters {
register(new SqlShowProcedureConverter());
registerCatalogConverters();
+ registerConnectionConverters();
registerMaterializedTableConverters();
registerModelConverters();
registerTableConverters();
@@ -138,6 +139,10 @@ public class SqlNodeConverters {
register(new SqlShowCreateCatalogConverter());
}
+ private static void registerConnectionConverters() {
+ register(new SqlCreateConnectionConverter());
+ }
+
private static void registerMaterializedTableConverters() {
register(new SqlAlterMaterializedTableAddDistributionConverter());
register(new SqlAlterMaterializedTableAddSchemaConverter());
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
new file mode 100644
index 00000000000..a410c95ec0b
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlConnectionOperationConverterTest.java
@@ -0,0 +1,124 @@
+/*
+ * 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;
+
+import org.apache.flink.sql.parser.error.SqlValidateException;
+import org.apache.flink.table.api.SqlParserException;
+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.ddl.CreateConnectionOperation;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Test for converting connection statements to operations. */
+class SqlConnectionOperationConverterTest extends
SqlNodeToOperationConversionTestBase {
+
+ @Test
+ void testCreateConnection() {
+ Operation operation = parse("CREATE CONNECTION my_conn WITH ('k' =
'v')");
+ assertThat(operation).isInstanceOf(CreateConnectionOperation.class);
+ CreateConnectionOperation op = (CreateConnectionOperation) operation;
+
+ assertThat(op.getConnectionIdentifier())
+ .isEqualTo(ObjectIdentifier.of("builtin", "default",
"my_conn"));
+
assertThat(op.getSensitiveConnection().getOptions()).isEqualTo(Map.of("k",
"v"));
+ assertThat(op.getSensitiveConnection().getComment()).isNull();
+ assertThat(op.isIgnoreIfExists()).isFalse();
+ assertThat(op.isTemporary()).isFalse();
+ }
+
+ @Test
+ void testCreateConnectionIfNotExists() {
+ Operation operation = parse("CREATE CONNECTION IF NOT EXISTS my_conn
WITH ('k' = 'v')");
+ CreateConnectionOperation op = (CreateConnectionOperation) operation;
+ assertThat(op.isIgnoreIfExists()).isTrue();
+ assertThat(op.isTemporary()).isFalse();
+ }
+
+ @Test
+ void testCreateTemporaryConnection() {
+ Operation operation = parse("CREATE TEMPORARY CONNECTION my_conn WITH
('k' = 'v')");
+ CreateConnectionOperation op = (CreateConnectionOperation) operation;
+ assertThat(op.isTemporary()).isTrue();
+ assertThat(op.isIgnoreIfExists()).isFalse();
+ }
+
+ @Test
+ void testCreateTemporarySystemConnection() {
+ Operation operation = parse("CREATE TEMPORARY SYSTEM CONNECTION
my_conn WITH ('k' = 'v')");
+ CreateConnectionOperation op = (CreateConnectionOperation) operation;
+ assertThat(op.isTemporary()).isTrue();
+ }
+
+ @Test
+ void testCreateConnectionWithComment() {
+ Operation operation =
+ parse("CREATE CONNECTION my_conn COMMENT 'hi there' WITH ('k'
= 'v')");
+ CreateConnectionOperation op = (CreateConnectionOperation) operation;
+ SensitiveConnection conn = op.getSensitiveConnection();
+ assertThat(conn.getComment()).isEqualTo("hi there");
+ }
+
+ @Test
+ void testCreateConnectionWithFullyQualifiedName() {
+ Operation operation = parse("CREATE CONNECTION cat1.db1.my_conn WITH
('k' = 'v')");
+ CreateConnectionOperation op = (CreateConnectionOperation) operation;
+ assertThat(op.getConnectionIdentifier())
+ .isEqualTo(ObjectIdentifier.of("cat1", "db1", "my_conn"));
+ }
+
+ @Test
+ void testCreateConnectionOptions() {
+ Operation operation =
+ parse("CREATE CONNECTION my_conn WITH ('k1' = 'v1', 'k2' =
'v2', 'k3' = 'v3')");
+ CreateConnectionOperation op = (CreateConnectionOperation) operation;
+ assertThat(op.getSensitiveConnection().getOptions())
+ .isEqualTo(Map.of("k1", "v1", "k2", "v2", "k3", "v3"));
+ }
+
+ @Test
+ void testAsSummaryStringMasksOptionValues() {
+ Operation operation =
+ parse(
+ "CREATE CONNECTION my_conn WITH ('user' = 'alice',
'password' = 'super-secret')");
+ String summary = operation.asSummaryString();
+
assertThat(summary).contains("user").contains("password").contains("****");
+
assertThat(summary).doesNotContain("alice").doesNotContain("super-secret");
+ }
+
+ @Test
+ void testCreateSystemConnectionWithoutTemporaryRejected() {
+ assertThatThrownBy(() -> parse("CREATE SYSTEM CONNECTION my_conn WITH
('k' = 'v')"))
+ .isInstanceOf(SqlParserException.class)
+ .hasMessageContaining("CREATE SYSTEM CONNECTION is not
supported");
+ }
+
+ @Test
+ void testCreateConnectionWithEmptyOptionsRejected() {
+ assertThatThrownBy(() -> parse("CREATE CONNECTION my_conn WITH ()"))
+ .isInstanceOf(SqlValidateException.class)
+ .hasMessageContaining("Connection property list can not be
empty.");
+ }
+}
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
new file mode 100644
index 00000000000..a9d99cbfe5d
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/CreateConnectionITCase.java
@@ -0,0 +1,87 @@
+/*
+ * 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.runtime.batch.sql;
+
+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.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.Assertions.entry;
+
+/** IT case for CREATE CONNECTION statement. */
+class CreateConnectionITCase extends BatchTestBase {
+
+ @Test
+ void testCreateTemporaryConnection() {
+ tEnv().executeSql(
+ "CREATE TEMPORARY CONNECTION my_conn COMMENT 'hi
there' "
+ + "WITH ('k' = 'v')");
+
+
assertThat(catalogManager().getConnection(connectionIdentifier("my_conn")))
+ .hasValueSatisfying(
+ connection -> {
+
assertThat(connection.getOptions()).containsOnly(entry("k", "v"));
+ assertThat(connection.getComment()).isEqualTo("hi
there");
+ });
+ }
+
+ @Test
+ void testCreateTemporaryConnectionRejectsDuplicate() {
+ tEnv().executeSql("CREATE TEMPORARY CONNECTION my_conn WITH ('k' =
'v1')");
+
+ assertThatThrownBy(
+ () ->
+ tEnv().executeSql(
+ "CREATE TEMPORARY CONNECTION
my_conn WITH ('k' = 'v2')"))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("Temporary connection");
+
+ tEnv().executeSql("CREATE TEMPORARY CONNECTION IF NOT EXISTS my_conn
WITH ('k' = 'v2')");
+
+
assertThat(catalogManager().getConnection(connectionIdentifier("my_conn")))
+ .hasValueSatisfying(
+ connection ->
+
assertThat(connection.getOptions()).containsOnly(entry("k", "v1")));
+ }
+
+ @Test
+ void testCreatePermanentConnectionRejectedWithoutSecretStore() {
+ assertThatThrownBy(() -> tEnv().executeSql("CREATE CONNECTION my_conn
WITH ('k' = 'v')"))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("WritableSecretStore must be
configured");
+ }
+
+ private CatalogManager catalogManager() {
+ return ((TableEnvironmentInternal) tEnv()).getCatalogManager();
+ }
+
+ private ObjectIdentifier connectionIdentifier(String connectionName) {
+ CatalogManager catalogManager = catalogManager();
+ return ObjectIdentifier.of(
+ catalogManager.getCurrentCatalog(),
+ catalogManager.getCurrentDatabase(),
+ connectionName);
+ }
+}
