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 f84ba2d8714 [FLINK-39889][table] Thread USING CONNECTION clause into
CatalogTable and CreateTableOperation (#28385)
f84ba2d8714 is described below
commit f84ba2d871429f776a5a6ed9a3c3580fa50fcc17
Author: Hao Li <[email protected]>
AuthorDate: Tue Aug 25 08:47:13 2026 -0700
[FLINK-39889][table] Thread USING CONNECTION clause into CatalogTable and
CreateTableOperation (#28385)
* [FLINK-39889][table] Convert table sqlnode to operation
Thread the connection identifier from a parsed USING CONNECTION clause
in SqlCreateTable through to CatalogTable and the CreateTableOperation.
- Add Optional<UnresolvedIdentifier> getConnection() to CatalogTable, a
connection() builder method, and the field/accessor in
DefaultCatalogTable (equals/hashCode/toString/copy updated).
- Delegate getConnection() in ResolvedCatalogTable.
- Read SqlCreateTable.getConnection() in AbstractCreateTableConverter and
set it on the built CatalogTable.
- Serialize/deserialize the connection identifier in CatalogPropertiesUtil
under a reserved 'connection.identifier' key, excluded from connector
options, so it round-trips and appears in CreateTableOperation summaries.
- Add a withConnection() OperationMatcher and tests covering conversion and
property serde.
Part of FLIP-529.
* [FLINK-39889][table] Address review: assert connection directly,
uppercase SQL keywords
* [FLINK-39889][table] Address review: add tests for malformed connection
names
Covers parser rejection of malformed compound identifiers, too-many-parts
and whitespace-only identifier segments failing validation, and unicode
connection names being accepted.
* [FLINK-39889][table] Address review: add multibyte connection name test
---
.../flink/table/catalog/CatalogPropertiesUtil.java | 39 +++++++++++++
.../apache/flink/table/catalog/CatalogTable.java | 16 +++++-
.../flink/table/catalog/DefaultCatalogTable.java | 30 ++++++++--
.../flink/table/catalog/ResolvedCatalogTable.java | 5 ++
.../table/catalog/CatalogPropertiesUtilTest.java | 7 +++
.../table/AbstractCreateTableConverter.java | 6 ++
.../operations/SqlDdlToOperationConverterTest.java | 66 ++++++++++++++++++++++
7 files changed, 164 insertions(+), 5 deletions(-)
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/CatalogPropertiesUtil.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/CatalogPropertiesUtil.java
index 58c8d76d858..fefd946f1d6 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/CatalogPropertiesUtil.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/CatalogPropertiesUtil.java
@@ -96,6 +96,9 @@ public final class CatalogPropertiesUtil {
final Optional<TableDistribution> distribution =
resolvedTable.getDistribution();
distribution.ifPresent(d -> serializeTableDistribution(properties,
d));
+ final Optional<UnresolvedIdentifier> connection =
resolvedTable.getConnection();
+ connection.ifPresent(c -> serializeConnection(properties, c));
+
properties.putAll(resolvedTable.getOptions());
properties.remove(IS_GENERIC); // reserved option
@@ -251,6 +254,8 @@ public final class CatalogPropertiesUtil {
final @Nullable TableDistribution distribution =
deserializeTableDistribution(properties);
+ final @Nullable UnresolvedIdentifier connection =
deserializeConnection(properties);
+
return CatalogTable.newBuilder()
.schema(schema)
.comment(comment)
@@ -258,6 +263,7 @@ public final class CatalogPropertiesUtil {
.distribution(distribution)
.options(options)
.snapshot(snapshot)
+ .connection(connection)
.build();
} catch (Exception e) {
throw new CatalogException("Error in deserializing catalog
table.", e);
@@ -445,12 +451,17 @@ public final class CatalogPropertiesUtil {
private static final String DISTRIBUTION_KEYS = compoundKey(DISTRIBUTION,
KEYS);
+ private static final String CONNECTION = "connection";
+
+ private static final String CONNECTION_IDENTIFIER =
compoundKey(CONNECTION, "identifier");
+
private static Map<String, String> deserializeOptions(Map<String, String>
map) {
return map.entrySet().stream()
.filter(
e -> {
final String key = e.getKey();
return !key.startsWith(DISTRIBUTION + SEPARATOR)
+ && !key.startsWith(CONNECTION + SEPARATOR)
&& !key.startsWith(PARTITION_KEYS +
SEPARATOR)
&& !key.startsWith(SCHEMA)
&& !key.equals(COMMENT)
@@ -652,6 +663,34 @@ public final class CatalogPropertiesUtil {
.collect(Collectors.toList()));
}
+ private static void serializeConnection(
+ Map<String, String> map, UnresolvedIdentifier connection) {
+ final List<String> parts = new ArrayList<>();
+ connection.getCatalogName().ifPresent(parts::add);
+ connection.getDatabaseName().ifPresent(parts::add);
+ parts.add(connection.getObjectName());
+
+ putIndexedProperties(
+ map,
+ CONNECTION_IDENTIFIER,
+ Collections.singletonList(NAME),
+
parts.stream().map(Collections::singletonList).collect(Collectors.toList()));
+ }
+
+ private static UnresolvedIdentifier deserializeConnection(Map<String,
String> map) {
+ final List<String> parts = new ArrayList<>();
+ int i = 0;
+ String partKey = compoundKey(CONNECTION_IDENTIFIER, i, NAME);
+ while (map.containsKey(partKey)) {
+ parts.add(getValue(map, partKey));
+ partKey = compoundKey(CONNECTION_IDENTIFIER, ++i, NAME);
+ }
+ if (parts.isEmpty()) {
+ return null;
+ }
+ return UnresolvedIdentifier.of(parts);
+ }
+
private static void serializeResolvedModelSchema(
Map<String, String> map,
ResolvedSchema inputSchema,
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/CatalogTable.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/CatalogTable.java
index c7539cc556e..3a234e9d1ea 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/CatalogTable.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/CatalogTable.java
@@ -109,6 +109,14 @@ public interface CatalogTable extends CatalogBaseTable {
return Optional.empty();
}
+ /**
+ * Returns the identifier of the connection that this table uses if the
{@code USING CONNECTION}
+ * clause is defined.
+ */
+ default Optional<UnresolvedIdentifier> getConnection() {
+ return Optional.empty();
+ }
+
//
--------------------------------------------------------------------------------------------
/** Builder for configuring and creating instances of {@link
CatalogTable}. */
@@ -120,6 +128,7 @@ public interface CatalogTable extends CatalogBaseTable {
private Map<String, String> options = Collections.emptyMap();
private @Nullable Long snapshot;
private @Nullable TableDistribution distribution;
+ private @Nullable UnresolvedIdentifier connection;
private Builder() {}
@@ -154,9 +163,14 @@ public interface CatalogTable extends CatalogBaseTable {
return this;
}
+ public Builder connection(@Nullable UnresolvedIdentifier connection) {
+ this.connection = connection;
+ return this;
+ }
+
public CatalogTable build() {
return new DefaultCatalogTable(
- schema, comment, partitionKeys, options, snapshot,
distribution);
+ schema, comment, partitionKeys, options, snapshot,
distribution, connection);
}
}
}
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/DefaultCatalogTable.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/DefaultCatalogTable.java
index f94c2399acc..b5e53959881 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/DefaultCatalogTable.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/DefaultCatalogTable.java
@@ -42,6 +42,7 @@ public class DefaultCatalogTable implements CatalogTable {
private final List<String> partitionKeys;
private final Map<String, String> options;
private final @Nullable Long snapshot;
+ private final @Nullable UnresolvedIdentifier connection;
protected DefaultCatalogTable(
Schema schema,
@@ -58,12 +59,24 @@ public class DefaultCatalogTable implements CatalogTable {
Map<String, String> options,
@Nullable Long snapshot,
@Nullable TableDistribution distribution) {
+ this(schema, comment, partitionKeys, options, snapshot, distribution,
null);
+ }
+
+ protected DefaultCatalogTable(
+ Schema schema,
+ @Nullable String comment,
+ List<String> partitionKeys,
+ Map<String, String> options,
+ @Nullable Long snapshot,
+ @Nullable TableDistribution distribution,
+ @Nullable UnresolvedIdentifier connection) {
this.schema = checkNotNull(schema, "Schema must not be null.");
this.comment = comment;
this.partitionKeys = checkNotNull(partitionKeys, "Partition keys must
not be null.");
this.options = checkNotNull(options, "Options must not be null.");
this.snapshot = snapshot;
this.distribution = distribution;
+ this.connection = connection;
checkArgument(
options.entrySet().stream()
@@ -96,6 +109,11 @@ public class DefaultCatalogTable implements CatalogTable {
return Optional.ofNullable(distribution);
}
+ @Override
+ public Optional<UnresolvedIdentifier> getConnection() {
+ return Optional.ofNullable(connection);
+ }
+
@Override
public Map<String, String> getOptions() {
return options;
@@ -109,13 +127,13 @@ public class DefaultCatalogTable implements CatalogTable {
@Override
public CatalogBaseTable copy() {
return new DefaultCatalogTable(
- schema, comment, partitionKeys, options, snapshot,
distribution);
+ schema, comment, partitionKeys, options, snapshot,
distribution, connection);
}
@Override
public CatalogTable copy(Map<String, String> options) {
return new DefaultCatalogTable(
- schema, comment, partitionKeys, options, snapshot,
distribution);
+ schema, comment, partitionKeys, options, snapshot,
distribution, connection);
}
@Override
@@ -142,12 +160,14 @@ public class DefaultCatalogTable implements CatalogTable {
&& Objects.equals(distribution, that.distribution)
&& partitionKeys.equals(that.partitionKeys)
&& options.equals(that.options)
- && Objects.equals(snapshot, that.snapshot);
+ && Objects.equals(snapshot, that.snapshot)
+ && Objects.equals(connection, that.connection);
}
@Override
public int hashCode() {
- return Objects.hash(schema, comment, distribution, partitionKeys,
options, snapshot);
+ return Objects.hash(
+ schema, comment, distribution, partitionKeys, options,
snapshot, connection);
}
@Override
@@ -166,6 +186,8 @@ public class DefaultCatalogTable implements CatalogTable {
+ ConfigurationUtils.hideSensitiveValues(options, List.of())
+ ", snapshot="
+ snapshot
+ + ", connection="
+ + connection
+ '}';
}
}
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/ResolvedCatalogTable.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/ResolvedCatalogTable.java
index f83b2601eb7..bcc6a378d23 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/ResolvedCatalogTable.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/ResolvedCatalogTable.java
@@ -145,6 +145,11 @@ public final class ResolvedCatalogTable
return origin.getDistribution();
}
+ @Override
+ public Optional<UnresolvedIdentifier> getConnection() {
+ return origin.getConnection();
+ }
+
@Override
public ResolvedCatalogTable copy(Map<String, String> options) {
return new ResolvedCatalogTable(origin.copy(options), resolvedSchema);
diff --git
a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/catalog/CatalogPropertiesUtilTest.java
b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/catalog/CatalogPropertiesUtilTest.java
index 3712ff2057f..3ffc27c4b5d 100644
---
a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/catalog/CatalogPropertiesUtilTest.java
+++
b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/catalog/CatalogPropertiesUtilTest.java
@@ -160,6 +160,13 @@ class CatalogPropertiesUtilTest {
.distribution(rangeDist)
.build(),
resolvedSchema),
+ new ResolvedCatalogTable(
+ CatalogTable.newBuilder()
+ .schema(schema)
+ .comment("some comment")
+ .connection(UnresolvedIdentifier.of("mycat",
"mydb", "myconn"))
+ .build(),
+ resolvedSchema),
new ResolvedCatalogMaterializedTable(
CatalogMaterializedTable.newBuilder()
.schema(schema)
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/table/AbstractCreateTableConverter.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/table/AbstractCreateTableConverter.java
index 5899c767f38..375166f60be 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/table/AbstractCreateTableConverter.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/table/AbstractCreateTableConverter.java
@@ -73,6 +73,11 @@ public abstract class AbstractCreateTableConverter<T extends
SqlCreateTable>
final Map<String, String> tableOptions =
mergeContext.getMergedTableOptions();
final TableDistribution distribution =
mergeContext.getMergedTableDistribution().orElse(null);
+ final UnresolvedIdentifier connection =
+ sqlCreateTable
+ .getConnection()
+ .map(c -> UnresolvedIdentifier.of(c.names))
+ .orElse(null);
final String comment = sqlCreateTable.getComment();
final CatalogTable catalogTable =
CatalogTable.newBuilder()
@@ -81,6 +86,7 @@ public abstract class AbstractCreateTableConverter<T extends
SqlCreateTable>
.distribution(distribution)
.options(tableOptions)
.partitionKeys(partitionKeys)
+ .connection(connection)
.build();
return context.getCatalogManager().resolveCatalogTable(catalogTable);
}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlDdlToOperationConverterTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlDdlToOperationConverterTest.java
index 2c5dd5d0d71..91de5b2aa89 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlDdlToOperationConverterTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlDdlToOperationConverterTest.java
@@ -24,6 +24,7 @@ import org.apache.flink.sql.parser.error.SqlValidateException;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.SqlDialect;
+import org.apache.flink.table.api.SqlParserException;
import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.catalog.Catalog;
import org.apache.flink.table.catalog.CatalogDatabaseImpl;
@@ -45,6 +46,7 @@ import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.catalog.TableChange;
import org.apache.flink.table.catalog.TableDistribution;
import org.apache.flink.table.catalog.TableDistribution.Kind;
+import org.apache.flink.table.catalog.UnresolvedIdentifier;
import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
import org.apache.flink.table.catalog.exceptions.FunctionAlreadyExistException;
import org.apache.flink.table.expressions.DefaultSqlFactory;
@@ -89,6 +91,7 @@ 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 org.junit.jupiter.params.provider.ValueSource;
import javax.annotation.Nullable;
@@ -658,6 +661,69 @@ class SqlDdlToOperationConverterTest extends
SqlNodeToOperationConversionTestBas
Collections.singletonList("a"), null)))));
}
+ @Test
+ void testCreateTableWithConnection() {
+ final String sql =
+ "CREATE TABLE derivedTable(\n"
+ + " a INT\n"
+ + ")\n"
+ + "USING CONNECTION mycat.mydb.myconn";
+ Operation operation = parseAndConvert(sql);
+ assertThat(operation).isInstanceOf(CreateTableOperation.class);
+ CreateTableOperation op = (CreateTableOperation) operation;
+ assertThat(op.getCatalogTable().getConnection())
+ .hasValue(UnresolvedIdentifier.of("mycat", "mydb", "myconn"));
+ }
+
+ @Test
+ void testCreateTableWithoutConnection() {
+ final String sql = "CREATE TABLE derivedTable(\n" + " a INT\n" + ")";
+ Operation operation = parseAndConvert(sql);
+ assertThat(operation).isInstanceOf(CreateTableOperation.class);
+ CreateTableOperation op = (CreateTableOperation) operation;
+ assertThat(op.getCatalogTable().getConnection()).isEmpty();
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"mycat....mydb....myconn", ".", "...", ".2.2."})
+ void testCreateTableWithMalformedConnectionNameFailsToParse(String
connectionName) {
+ final String sql = "CREATE TABLE derivedTable(a INT) USING CONNECTION
" + connectionName;
+ assertThatThrownBy(() ->
parseAndConvert(sql)).isInstanceOf(SqlParserException.class);
+ }
+
+ @Test
+ void testCreateTableWithTooManyConnectionNameParts() {
+ final String sql =
+ "CREATE TABLE derivedTable(a INT) USING CONNECTION
mycat.mydb.mygroup.myconn";
+ assertThatThrownBy(() -> parseAndConvert(sql))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("Object identifier must consist of 1 to
3 parts.");
+ }
+
+ @Test
+ void testCreateTableWithWhitespaceOnlyConnectionNamePart() {
+ final String sql = "CREATE TABLE derivedTable(a INT) USING CONNECTION
mycat.` `.myconn";
+ assertThatThrownBy(() -> parseAndConvert(sql))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining(
+ "Parts of the object identifier are null or
whitespace-only.");
+ }
+
+ @Test
+ void testCreateTableWithUnicodeConnectionName() {
+ final String sql = "CREATE TABLE derivedTable(a INT) USING CONNECTION
`😍.😍`";
+ CreateTableOperation op = (CreateTableOperation) parseAndConvert(sql);
+
assertThat(op.getCatalogTable().getConnection()).hasValue(UnresolvedIdentifier.of("😍.😍"));
+ }
+
+ @Test
+ void testCreateTableWithMultibyteConnectionName() {
+ final String sql = "CREATE TABLE derivedTable(a INT) USING CONNECTION
`目录`.`Привет`.`café`";
+ CreateTableOperation op = (CreateTableOperation) parseAndConvert(sql);
+ assertThat(op.getCatalogTable().getConnection())
+ .hasValue(UnresolvedIdentifier.of("目录", "Привет", "café"));
+ }
+
@Test
void testCreateTableInvalidDistribution() {
final String sql =
