This is an automated email from the ASF dual-hosted git repository.
yuqi1129 pushed a commit to branch branch-1.3
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/branch-1.3 by this push:
new 7a58dd7132 [Cherry-pick to branch-1.3] [#13945] fix(jdbc): Honor pool
wait settings and validate MySQL table properties (#13946) (#14035)
7a58dd7132 is described below
commit 7a58dd7132ff121a993481d8b0d2169e8c8cfab7
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Wed Oct 21 14:37:52 2026 +0800
[Cherry-pick to branch-1.3] [#13945] fix(jdbc): Honor pool wait settings
and validate MySQL table properties (#13946) (#14035)
### What changes were proposed in this pull request?
Backport commit `3ce91977a2d137af39321716a9fb5813922afceb` (#13946) to
`branch-1.3`: honor JDBC pool wait settings and validate MySQL table
properties.
Resolve the committed conflict markers while retaining the 1.3
validation query, connection timeout handling and Spark
`getPGInstance()` API. Adapt tests to the existing 1.3 ALTER connection
lifecycle. Pool max-idle and table credential-vending settings are not
introduced by this backport.
### Why are the changes needed?
Configured pool wait values could be lost or overwritten, and
unsupported MySQL table options could reach generated SQL. This backport
brings the fixes to 1.3.
### Does this PR introduce _any_ user-facing change?
Yes. JDBC pool wait settings are honored, and unsupported or invalid
MySQL table options are rejected before DDL is executed.
### How was this patch tested?
- 70 targeted non-Docker tests passed across JDBC common, MySQL, Flink
common and Spark common.
- Formatting and conflict-marker checks passed.
- Six mutation cases fail if the retained `SELECT 1` pool validation
query is removed, then pass after restoration.
- Docker tests were not run locally because the Docker daemon is
unavailable; CI validation is pending.
---------
Co-authored-by: Qi Yu <[email protected]>
---
.../jdbc/JdbcCatalogPropertiesMetadata.java | 7 +
.../catalog/jdbc/utils/DataSourceUtils.java | 7 +-
.../jdbc/TestJdbcCatalogPropertiesMetadata.java | 91 +++++++++
.../mysql/MysqlTablePropertiesMetadata.java | 34 +++-
.../mysql/operation/MysqlTableOperations.java | 22 ++-
.../mysql/TestMysqlTablePropertiesMetadata.java | 214 +++++++++++++++++++++
.../mysql/operation/TestMysqlTableOperations.java | 11 +-
.../TestMysqlTableOperationsSqlGeneration.java | 92 +++++++++
docs/jdbc-clickhouse-catalog.md | 2 +-
docs/jdbc-doris-catalog.md | 2 +-
docs/jdbc-mysql-catalog.md | 6 +-
docs/jdbc-oceanbase-catalog.md | 2 +-
docs/jdbc-postgresql-catalog.md | 2 +-
docs/jdbc-starrocks-catalog.md | 2 +-
.../connector/jdbc/JdbcPropertiesConverter.java | 21 ++
.../connector/integration/test/FlinkCommonIT.java | 35 ++--
.../test/jdbc/FlinkJdbcMysqlCatalogIT.java | 32 +++
.../AbstractJdbcPropertiesConverterTestSuite.java | 21 ++
.../connector/jdbc/JdbcPropertiesConverter.java | 6 +-
.../jdbc/TestJdbcPropertiesConverter.java | 42 ++++
20 files changed, 611 insertions(+), 40 deletions(-)
diff --git
a/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/JdbcCatalogPropertiesMetadata.java
b/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/JdbcCatalogPropertiesMetadata.java
index 5f6c84ba84..5dc879452e 100644
---
a/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/JdbcCatalogPropertiesMetadata.java
+++
b/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/JdbcCatalogPropertiesMetadata.java
@@ -46,6 +46,7 @@ public class JdbcCatalogPropertiesMetadata extends
BaseCatalogPropertiesMetadata
JdbcConfig.PASSWORD.getKey(),
JdbcConfig.POOL_MIN_SIZE.getKey(),
JdbcConfig.POOL_MAX_SIZE.getKey(),
+ JdbcConfig.POOL_MAX_WAIT_MS.getKey(),
JdbcConfig.TEST_ON_BORROW.getKey(),
JdbcConfig.CONNECT_TIMEOUT_MS.getKey(),
JdbcConfig.VERIFY_ON_CREATE.getKey());
@@ -107,6 +108,12 @@ public class JdbcCatalogPropertiesMetadata extends
BaseCatalogPropertiesMetadata
JdbcConfig.POOL_MAX_SIZE.getDefaultValue(),
false /* hidden */,
false /* reserved */),
+ longOptionalPropertyEntry(
+ JdbcConfig.POOL_MAX_WAIT_MS.getKey(),
+ JdbcConfig.POOL_MAX_WAIT_MS.getDoc(),
+ false /* immutable */,
+ JdbcConfig.POOL_MAX_WAIT_MS.getDefaultValue(),
+ false /* hidden */),
booleanPropertyEntry(
JdbcConfig.TEST_ON_BORROW.getKey(),
JdbcConfig.TEST_ON_BORROW.getDoc(),
diff --git
a/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/utils/DataSourceUtils.java
b/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/utils/DataSourceUtils.java
index 1430cf4c8c..c298fbd4d1 100644
---
a/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/utils/DataSourceUtils.java
+++
b/catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/utils/DataSourceUtils.java
@@ -122,7 +122,12 @@ public class DataSourceUtils {
// executed to confirm whether the connection is valid.
basicDataSource.setTestOnBorrow(jdbcConfig.getTestOnBorrow());
basicDataSource.setValidationQuery(POOL_TEST_QUERY);
- basicDataSource.setMaxWait(Duration.ofMillis(jdbcConfig.getMaxWaitMs()));
+ // Keep a maxWaitMillis supplied through the DBCP bypass unless the
canonical setting is
+ // explicitly configured.
+ if
(jdbcConfig.getAllConfig().containsKey(JdbcConfig.POOL_MAX_WAIT_MS.getKey())
+ || !jdbcConfig.getAllConfig().containsKey("maxWaitMillis")) {
+ basicDataSource.setMaxWait(Duration.ofMillis(jdbcConfig.getMaxWaitMs()));
+ }
// Apply the driver-aware connect timeout as a DBCP connection property
(not a pool key, so the
// pool stays lazy). It bounds a single connection and takes effect on the
first real connection
// — i.e. create-time verification — so an unreachable backend fails fast
instead of hanging.
diff --git
a/catalogs/catalog-jdbc-common/src/test/java/org/apache/gravitino/catalog/jdbc/TestJdbcCatalogPropertiesMetadata.java
b/catalogs/catalog-jdbc-common/src/test/java/org/apache/gravitino/catalog/jdbc/TestJdbcCatalogPropertiesMetadata.java
index d810dd1d72..a77a6be771 100644
---
a/catalogs/catalog-jdbc-common/src/test/java/org/apache/gravitino/catalog/jdbc/TestJdbcCatalogPropertiesMetadata.java
+++
b/catalogs/catalog-jdbc-common/src/test/java/org/apache/gravitino/catalog/jdbc/TestJdbcCatalogPropertiesMetadata.java
@@ -20,22 +20,31 @@ package org.apache.gravitino.catalog.jdbc;
import com.google.common.collect.ImmutableMap;
import java.time.Instant;
+import java.util.HashMap;
import java.util.Map;
+import org.apache.commons.dbcp2.BasicDataSource;
+import org.apache.commons.lang3.reflect.FieldUtils;
import org.apache.gravitino.Catalog;
import org.apache.gravitino.Namespace;
import org.apache.gravitino.catalog.jdbc.config.JdbcConfig;
import
org.apache.gravitino.catalog.jdbc.converter.JdbcColumnDefaultValueConverter;
import org.apache.gravitino.catalog.jdbc.converter.JdbcTypeConverter;
import
org.apache.gravitino.catalog.jdbc.converter.SqliteColumnDefaultValueConverter;
+import org.apache.gravitino.catalog.jdbc.converter.SqliteExceptionConverter;
import org.apache.gravitino.catalog.jdbc.converter.SqliteTypeConverter;
import org.apache.gravitino.catalog.jdbc.operation.JdbcDatabaseOperations;
import org.apache.gravitino.catalog.jdbc.operation.JdbcTableOperations;
import org.apache.gravitino.catalog.jdbc.operation.SqliteDatabaseOperations;
import org.apache.gravitino.catalog.jdbc.operation.SqliteTableOperations;
+import org.apache.gravitino.connector.CatalogInfo;
+import org.apache.gravitino.exceptions.GravitinoRuntimeException;
import org.apache.gravitino.meta.AuditInfo;
import org.apache.gravitino.meta.CatalogEntity;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
public class TestJdbcCatalogPropertiesMetadata {
@@ -61,6 +70,88 @@ public class TestJdbcCatalogPropertiesMetadata {
.shouldValidateOnCreate());
}
+ @ParameterizedTest
+ @CsvSource({
+ "none,none,none,30000",
+ "1250,none,none,1250",
+ "none,2300,none,2300",
+ "1250,2300,none,1250",
+ "none,none,3400,3400",
+ "1250,none,3400,1250"
+ })
+ void testPoolMaxWaitReachesCatalogDataSource(
+ String canonical, String bypass, String dbcpBypass, long expected)
throws Exception {
+ Map<String, String> properties = poolWaitProperties();
+ if (!"none".equals(canonical)) {
+ properties.put(JdbcConfig.POOL_MAX_WAIT_MS.getKey(), canonical);
+ }
+ if (!"none".equals(bypass)) {
+ properties.put("gravitino.bypass." +
JdbcConfig.POOL_MAX_WAIT_MS.getKey(), bypass);
+ }
+ if (!"none".equals(dbcpBypass)) {
+ // The DBCP-native key documented as the bypass example in the JDBC
catalog docs.
+ properties.put("gravitino.bypass.maxWaitMillis", dbcpBypass);
+ }
+ JdbcCatalogPropertiesMetadata metadata = new
JdbcCatalogPropertiesMetadata();
+
Assertions.assertTrue(metadata.containsProperty(JdbcConfig.POOL_MAX_WAIT_MS.getKey()));
+
Assertions.assertFalse(metadata.isHiddenProperty(JdbcConfig.POOL_MAX_WAIT_MS.getKey()));
+
Assertions.assertFalse(metadata.isImmutableProperty(JdbcConfig.POOL_MAX_WAIT_MS.getKey()));
+ try (JdbcCatalogOperations operations = poolWaitOperations()) {
+ operations.initialize(properties, poolWaitCatalogInfo(properties), new
TestJdbcCatalog());
+ BasicDataSource pool = (BasicDataSource)
FieldUtils.readField(operations, "dataSource", true);
+ Assertions.assertEquals(expected, pool.getMaxWaitDuration().toMillis());
+ Assertions.assertEquals("SELECT 1", pool.getValidationQuery());
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"0", "-1", "not-a-number"})
+ void testInvalidPoolMaxWaitIsRejected(String value) {
+ Map<String, String> properties = poolWaitProperties();
+ properties.put(JdbcConfig.POOL_MAX_WAIT_MS.getKey(), value);
+ try (JdbcCatalogOperations operations = poolWaitOperations()) {
+ Assertions.assertThrows(
+ GravitinoRuntimeException.class,
+ () ->
+ operations.initialize(
+ properties, poolWaitCatalogInfo(properties), new
TestJdbcCatalog()));
+ }
+ }
+
+ private static Map<String, String> poolWaitProperties() {
+ return new HashMap<>(
+ Map.of(
+ JdbcConfig.JDBC_URL.getKey(),
+ "jdbc:sqlite::memory:",
+ JdbcConfig.JDBC_DRIVER.getKey(),
+ "org.sqlite.JDBC",
+ JdbcConfig.USERNAME.getKey(),
+ "user",
+ JdbcConfig.PASSWORD.getKey(),
+ "password"));
+ }
+
+ private static CatalogInfo poolWaitCatalogInfo(Map<String, String>
properties) {
+ return new CatalogInfo(
+ 1L,
+ "jdbc_catalog",
+ Catalog.Type.RELATIONAL,
+ "jdbc-test",
+ "test",
+ properties,
+
AuditInfo.builder().withCreator("test").withCreateTime(Instant.EPOCH).build(),
+ Namespace.of("metalake"));
+ }
+
+ private static JdbcCatalogOperations poolWaitOperations() {
+ return new JdbcCatalogOperations(
+ new SqliteExceptionConverter(),
+ new SqliteTypeConverter(),
+ new SqliteDatabaseOperations(":memory:"),
+ new SqliteTableOperations(),
+ new SqliteColumnDefaultValueConverter());
+ }
+
private static JdbcCatalog buildCatalog(Map<String, String> extraProperties)
{
Map<String, String> properties =
ImmutableMap.<String, String>builder()
diff --git
a/catalogs/catalog-jdbc-mysql/src/main/java/org/apache/gravitino/catalog/mysql/MysqlTablePropertiesMetadata.java
b/catalogs/catalog-jdbc-mysql/src/main/java/org/apache/gravitino/catalog/mysql/MysqlTablePropertiesMetadata.java
index e4de0a0cbd..472a10c6f7 100644
---
a/catalogs/catalog-jdbc-mysql/src/main/java/org/apache/gravitino/catalog/mysql/MysqlTablePropertiesMetadata.java
+++
b/catalogs/catalog-jdbc-mysql/src/main/java/org/apache/gravitino/catalog/mysql/MysqlTablePropertiesMetadata.java
@@ -18,10 +18,12 @@
*/
package org.apache.gravitino.catalog.mysql;
+import static
org.apache.gravitino.catalog.PropertiesMetadataHelpers.checkValueFormat;
import static
org.apache.gravitino.connector.PropertyEntry.enumImmutablePropertyEntry;
import static
org.apache.gravitino.connector.PropertyEntry.integerOptionalPropertyEntry;
import static
org.apache.gravitino.connector.PropertyEntry.stringReservedPropertyEntry;
+import com.google.common.base.Preconditions;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
@@ -107,17 +109,23 @@ public class MysqlTablePropertiesMetadata extends
JdbcTablePropertiesMetadata {
@Override
public Map<String, String> transformToJdbcProperties(Map<String, String>
properties) {
- return Collections.unmodifiableMap(
- new HashMap<String, String>() {
- {
- properties.forEach(
- (key, value) -> {
- if (GRAVITINO_CONFIG_TO_MYSQL.containsKey(key)) {
- put(GRAVITINO_CONFIG_TO_MYSQL.get(key), value);
- }
- });
+ Map<String, String> result = new HashMap<>();
+ properties.forEach(
+ (key, value) -> {
+ Preconditions.checkArgument(
+ containsProperty(key),
+ "Unsupported MySQL table property: %s. Supported properties: %s",
+ key,
+ String.join(", ", GRAVITINO_CONFIG_TO_MYSQL.keySet()));
+ if (GRAVITINO_CONFIG_TO_MYSQL.containsKey(key)) {
+ // A null value represents RemoveProperty in
JdbcCatalogOperations. Non-SQL metadata,
+ // such as the internal identifier, must not become options.
+ result.put(
+ GRAVITINO_CONFIG_TO_MYSQL.get(key),
+ value == null ? null :
validateAndEncode(getPropertyEntry(key), value));
}
});
+ return Collections.unmodifiableMap(result);
}
@Override
@@ -135,4 +143,12 @@ public class MysqlTablePropertiesMetadata extends
JdbcTablePropertiesMetadata {
}
});
}
+
+ private static <T> String validateAndEncode(PropertyEntry<T> entry, String
value) {
+ T decoded = checkValueFormat(entry.getName(), value, entry::decode);
+ if (decoded instanceof ENGINE) {
+ return ((ENGINE) decoded).getValue();
+ }
+ return entry.getEncoder().apply(decoded);
+ }
}
diff --git
a/catalogs/catalog-jdbc-mysql/src/main/java/org/apache/gravitino/catalog/mysql/operation/MysqlTableOperations.java
b/catalogs/catalog-jdbc-mysql/src/main/java/org/apache/gravitino/catalog/mysql/operation/MysqlTableOperations.java
index f50ccca7df..c3e9b7898d 100644
---
a/catalogs/catalog-jdbc-mysql/src/main/java/org/apache/gravitino/catalog/mysql/operation/MysqlTableOperations.java
+++
b/catalogs/catalog-jdbc-mysql/src/main/java/org/apache/gravitino/catalog/mysql/operation/MysqlTableOperations.java
@@ -48,6 +48,7 @@ import org.apache.gravitino.StringIdentifier;
import org.apache.gravitino.catalog.jdbc.JdbcColumn;
import org.apache.gravitino.catalog.jdbc.JdbcTable;
import org.apache.gravitino.catalog.jdbc.operation.JdbcTableOperations;
+import org.apache.gravitino.catalog.mysql.MysqlTablePropertiesMetadata;
import org.apache.gravitino.exceptions.NoSuchColumnException;
import org.apache.gravitino.exceptions.NoSuchTableException;
import org.apache.gravitino.rel.Column;
@@ -63,6 +64,8 @@ import org.apache.gravitino.rel.types.Types;
/** Table operations for MySQL. */
public class MysqlTableOperations extends JdbcTableOperations {
+ private static final MysqlTablePropertiesMetadata TABLE_PROPERTIES_METADATA =
+ new MysqlTablePropertiesMetadata();
private static final String BACK_QUOTE = "`";
private static final String MYSQL_AUTO_INCREMENT = "AUTO_INCREMENT";
private static final String MYSQL_NOT_SUPPORT_NESTED_COLUMN_MSG =
@@ -132,7 +135,7 @@ public class MysqlTableOperations extends
JdbcTableOperations {
if (MapUtils.isNotEmpty(properties)) {
sqlBuilder.append(
properties.entrySet().stream()
- .map(entry -> String.format("%s = %s", entry.getKey(),
entry.getValue()))
+ .map(entry -> generateTableProperty(entry.getKey(),
entry.getValue()))
.collect(Collectors.joining(",\n", "\n", "")));
}
@@ -503,11 +506,24 @@ public class MysqlTableOperations extends
JdbcTableOperations {
private String generateTableProperties(List<TableChange.SetProperty>
setProperties) {
return setProperties.stream()
.map(
- setProperty ->
- String.format("%s = %s", setProperty.getProperty(),
setProperty.getValue()))
+ setProperty -> generateTableProperty(setProperty.getProperty(),
setProperty.getValue()))
.collect(Collectors.joining(",\n"));
}
+ private String generateTableProperty(String key, String value) {
+ Map<String, String> logicalProperties =
+
TABLE_PROPERTIES_METADATA.convertFromJdbcProperties(Collections.singletonMap(key,
value));
+ Preconditions.checkArgument(
+ !logicalProperties.isEmpty(),
+ "Unsupported MySQL table option: %s. Supported options: %s",
+ key,
+ String.join(", ",
MysqlTablePropertiesMetadata.GRAVITINO_CONFIG_TO_MYSQL.values()));
+ String encoded =
+
TABLE_PROPERTIES_METADATA.transformToJdbcProperties(logicalProperties).get(key);
+ Preconditions.checkArgument(encoded != null, "MySQL table option %s cannot
be null", key);
+ return key + " = " + encoded;
+ }
+
private String updateColumnCommentFieldDefinition(
TableChange.UpdateColumnComment updateColumnComment, JdbcTable
jdbcTable) {
String newComment = updateColumnComment.getNewComment();
diff --git
a/catalogs/catalog-jdbc-mysql/src/test/java/org/apache/gravitino/catalog/mysql/TestMysqlTablePropertiesMetadata.java
b/catalogs/catalog-jdbc-mysql/src/test/java/org/apache/gravitino/catalog/mysql/TestMysqlTablePropertiesMetadata.java
new file mode 100644
index 0000000000..228358a3f8
--- /dev/null
+++
b/catalogs/catalog-jdbc-mysql/src/test/java/org/apache/gravitino/catalog/mysql/TestMysqlTablePropertiesMetadata.java
@@ -0,0 +1,214 @@
+/*
+ * 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.gravitino.catalog.mysql;
+
+import java.time.Instant;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.gravitino.Catalog;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.Namespace;
+import org.apache.gravitino.StringIdentifier;
+import org.apache.gravitino.catalog.PropertiesMetadataHelpers;
+import org.apache.gravitino.catalog.jdbc.JdbcCatalogOperations;
+import org.apache.gravitino.catalog.jdbc.JdbcColumn;
+import org.apache.gravitino.catalog.jdbc.JdbcTable;
+import org.apache.gravitino.catalog.jdbc.converter.JdbcExceptionConverter;
+import
org.apache.gravitino.catalog.mysql.converter.MysqlColumnDefaultValueConverter;
+import org.apache.gravitino.catalog.mysql.converter.MysqlTypeConverter;
+import org.apache.gravitino.catalog.mysql.operation.MysqlDatabaseOperations;
+import org.apache.gravitino.catalog.mysql.operation.MysqlTableOperations;
+import org.apache.gravitino.connector.CatalogInfo;
+import org.apache.gravitino.meta.AuditInfo;
+import org.apache.gravitino.rel.Column;
+import org.apache.gravitino.rel.Table;
+import org.apache.gravitino.rel.TableChange;
+import org.apache.gravitino.rel.expressions.distributions.Distribution;
+import org.apache.gravitino.rel.expressions.distributions.Distributions;
+import org.apache.gravitino.rel.expressions.sorts.SortOrder;
+import org.apache.gravitino.rel.expressions.transforms.Transform;
+import org.apache.gravitino.rel.expressions.transforms.Transforms;
+import org.apache.gravitino.rel.indexes.Index;
+import org.apache.gravitino.rel.indexes.Indexes;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+class TestMysqlTablePropertiesMetadata {
+ private static final NameIdentifier IDENT =
+ NameIdentifier.of("metalake", "catalog", "database", "table");
+
+ @ParameterizedTest
+ @ValueSource(strings = {"unknown-key", "ENGINE", "ROW_FORMAT", "engine; DROP
TABLE t"})
+ void testCatalogRejectsUnknownPropertiesBeforeCreateOrAlter(String key) {
+ RecordingTableOperations table = new RecordingTableOperations();
+ try (JdbcCatalogOperations operations = operations(table)) {
+ IllegalArgumentException create =
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () -> create(operations,
Map.of(key, "MEMORY")));
+ Assertions.assertTrue(create.getMessage().contains(key));
+ Assertions.assertTrue(
+ create.getMessage().contains("Supported properties:
auto-increment-offset, engine"),
+ create.getMessage());
+ IllegalArgumentException alter =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> operations.alterTable(IDENT, TableChange.setProperty(key,
"MEMORY")));
+ Assertions.assertTrue(alter.getMessage().contains(key));
+ IllegalArgumentException remove =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> operations.alterTable(IDENT,
TableChange.removeProperty(key)));
+ Assertions.assertTrue(remove.getMessage().contains(key));
+ Assertions.assertEquals(0, table.writes);
+ }
+ }
+
+ @ParameterizedTest
+ @CsvSource({"engine,InnoDB; DROP TABLE t", "auto-increment-offset,42; DROP
TABLE t"})
+ void testInvalidDeclaredValuesAreRejectedBeforeCreate(String key, String
value) {
+ RecordingTableOperations table = new RecordingTableOperations();
+ try (JdbcCatalogOperations operations = operations(table)) {
+ IllegalArgumentException error =
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () -> create(operations,
Map.of(key, value)));
+ Assertions.assertTrue(error.getMessage().contains(key));
+ Assertions.assertEquals(0, table.writes);
+ }
+ }
+
+ @Test
+ void testValidPropertiesRoundTripAndMetadataDoesNotBecomeSql() {
+ RecordingTableOperations table = new RecordingTableOperations();
+ try (JdbcCatalogOperations operations = operations(table)) {
+ Table created =
+ create(operations, Map.of("engine", "InNoDB",
"auto-increment-offset", "+0042"));
+ Assertions.assertEquals(Map.of("ENGINE", "InnoDB", "AUTO_INCREMENT",
"42"), table.properties);
+ Assertions.assertEquals(
+ Map.of("engine", "InnoDB", "auto-increment-offset", "42"),
created.properties());
+ Table loaded = operations.loadTable(IDENT);
+ Assertions.assertEquals("InnoDB", loaded.properties().get("engine"));
+ Assertions.assertEquals("42",
loaded.properties().get("auto-increment-offset"));
+ Assertions.assertEquals(1, table.writes);
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"engine", "auto-increment-offset"})
+ void testDeclaredPropertiesRemainImmutable(String key) {
+ IllegalArgumentException error =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ PropertiesMetadataHelpers.validatePropertyForAlter(
+ new MysqlTablePropertiesMetadata(), Map.of(key, "42"),
Map.of()));
+ Assertions.assertTrue(error.getMessage().contains("immutable"));
+ Map<String, String> removal = new HashMap<>();
+ removal.put(key, null);
+ Assertions.assertTrue(
+ new
MysqlTablePropertiesMetadata().transformToJdbcProperties(removal).containsValue(null));
+ }
+
+ private static Table create(JdbcCatalogOperations operations, Map<String,
String> properties) {
+ return operations.createTable(
+ IDENT,
+ new Column[0],
+ "comment",
+ StringIdentifier.newPropertiesWithId(StringIdentifier.fromId(1L),
properties),
+ Transforms.EMPTY_TRANSFORM,
+ Distributions.NONE,
+ new SortOrder[0],
+ Indexes.EMPTY_INDEXES);
+ }
+
+ private static JdbcCatalogOperations operations(RecordingTableOperations
table) {
+ JdbcCatalogOperations operations =
+ new JdbcCatalogOperations(
+ new JdbcExceptionConverter(),
+ new MysqlTypeConverter(),
+ new MysqlDatabaseOperations(),
+ table,
+ new MysqlColumnDefaultValueConverter());
+ Map<String, String> conf =
+ Map.of(
+ "jdbc-url",
+ "jdbc:mysql://127.0.0.1:1/unused",
+ "jdbc-driver",
+ "com.mysql.cj.jdbc.Driver",
+ "jdbc-user",
+ "test",
+ "jdbc-password",
+ "test");
+ CatalogInfo info =
+ new CatalogInfo(
+ 1L,
+ "catalog",
+ Catalog.Type.RELATIONAL,
+ "jdbc-mysql",
+ "test",
+ conf,
+
AuditInfo.builder().withCreator("test").withCreateTime(Instant.EPOCH).build(),
+ Namespace.of("metalake"));
+ operations.initialize(conf, info, new MysqlCatalog());
+ return operations;
+ }
+
+ private static class RecordingTableOperations extends MysqlTableOperations {
+ private int writes;
+ private Map<String, String> properties = Collections.emptyMap();
+ private String comment;
+
+ /** Records a create after catalog property validation and translation. */
+ @Override
+ public void create(
+ String databaseName,
+ String tableName,
+ JdbcColumn[] columns,
+ String comment,
+ Map<String, String> properties,
+ Transform[] partitioning,
+ Distribution distribution,
+ Index[] indexes,
+ SortOrder[] sortOrders) {
+ writes++;
+ this.properties = new HashMap<>(properties);
+ this.comment = comment;
+ }
+
+ /** Records alterations reaching the table operation boundary. */
+ @Override
+ public void alterTable(String databaseName, String tableName,
TableChange... changes) {
+ writes++;
+ }
+
+ /** Returns the state recorded by create. */
+ @Override
+ public JdbcTable load(String databaseName, String tableName) {
+ return JdbcTable.builder()
+ .withName(tableName)
+ .withColumns(new JdbcColumn[0])
+ .withProperties(properties)
+ .withComment(comment)
+ .build();
+ }
+ }
+}
diff --git
a/catalogs/catalog-jdbc-mysql/src/test/java/org/apache/gravitino/catalog/mysql/operation/TestMysqlTableOperations.java
b/catalogs/catalog-jdbc-mysql/src/test/java/org/apache/gravitino/catalog/mysql/operation/TestMysqlTableOperations.java
index b48e8a9e08..f9b38d6419 100644
---
a/catalogs/catalog-jdbc-mysql/src/test/java/org/apache/gravitino/catalog/mysql/operation/TestMysqlTableOperations.java
+++
b/catalogs/catalog-jdbc-mysql/src/test/java/org/apache/gravitino/catalog/mysql/operation/TestMysqlTableOperations.java
@@ -37,7 +37,6 @@ import org.apache.commons.lang3.StringUtils;
import org.apache.gravitino.catalog.jdbc.JdbcColumn;
import org.apache.gravitino.catalog.jdbc.JdbcTable;
import org.apache.gravitino.catalog.jdbc.utils.DataSourceUtils;
-import org.apache.gravitino.exceptions.GravitinoRuntimeException;
import org.apache.gravitino.rel.Column;
import org.apache.gravitino.rel.TableChange;
import org.apache.gravitino.rel.expressions.distributions.Distributions;
@@ -155,15 +154,13 @@ public class TestMysqlTableOperations extends TestMysql {
assertionsTableInfo(
newName, tableComment, alterColumns, properties, indexes,
Transforms.EMPTY_TRANSFORM, load);
- // Detect unsupported properties
+ // Reject an unsupported engine before generating or executing DDL.
TableChange setProperty = TableChange.setProperty(MYSQL_ENGINE_KEY, "ABC");
- GravitinoRuntimeException gravitinoRuntimeException =
+ IllegalArgumentException invalidEngine =
Assertions.assertThrows(
- GravitinoRuntimeException.class,
+ IllegalArgumentException.class,
() -> TABLE_OPERATIONS.alterTable(TEST_DB_NAME.toString(),
newName, setProperty));
- Assertions.assertTrue(
- StringUtils.contains(
- gravitinoRuntimeException.getMessage(), "Unknown storage engine
'ABC'"));
+ Assertions.assertTrue(invalidEngine.getMessage().contains("engine"));
// delete column
TABLE_OPERATIONS.alterTable(
diff --git
a/catalogs/catalog-jdbc-mysql/src/test/java/org/apache/gravitino/catalog/mysql/operation/TestMysqlTableOperationsSqlGeneration.java
b/catalogs/catalog-jdbc-mysql/src/test/java/org/apache/gravitino/catalog/mysql/operation/TestMysqlTableOperationsSqlGeneration.java
index ad8966c2c8..0b9f51a1a0 100644
---
a/catalogs/catalog-jdbc-mysql/src/test/java/org/apache/gravitino/catalog/mysql/operation/TestMysqlTableOperationsSqlGeneration.java
+++
b/catalogs/catalog-jdbc-mysql/src/test/java/org/apache/gravitino/catalog/mysql/operation/TestMysqlTableOperationsSqlGeneration.java
@@ -19,10 +19,13 @@
package org.apache.gravitino.catalog.mysql.operation;
import java.util.Collections;
+import java.util.Map;
+import java.util.stream.Stream;
import org.apache.gravitino.catalog.jdbc.JdbcColumn;
import org.apache.gravitino.catalog.jdbc.JdbcTable;
import
org.apache.gravitino.catalog.jdbc.converter.JdbcColumnDefaultValueConverter;
import org.apache.gravitino.catalog.jdbc.converter.JdbcExceptionConverter;
+import org.apache.gravitino.catalog.jdbc.utils.ConnectionCountingDataSource;
import org.apache.gravitino.catalog.mysql.converter.MysqlTypeConverter;
import org.apache.gravitino.rel.TableChange;
import org.apache.gravitino.rel.expressions.distributions.Distributions;
@@ -32,6 +35,9 @@ import org.apache.gravitino.rel.indexes.Indexes;
import org.apache.gravitino.rel.types.Types;
import org.junit.jupiter.api.Assertions;
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;
public class TestMysqlTableOperationsSqlGeneration {
@@ -42,6 +48,11 @@ public class TestMysqlTableOperationsSqlGeneration {
super.columnDefaultValueConverter = new
JdbcColumnDefaultValueConverter();
}
+ private TestableMysqlTableOperations(ConnectionCountingDataSource
connections) {
+ this();
+ dataSource = connections.dataSource();
+ }
+
public String createTableSql(String tableName, JdbcColumn[] columns) {
return createTableSql(tableName, columns, "comment");
}
@@ -68,6 +79,77 @@ public class TestMysqlTableOperationsSqlGeneration {
}
}
+ @ParameterizedTest
+ @MethodSource("invalidTableOptions")
+ void testInvalidOptionsAreRejectedBeforeCreateAndAlterSql(String key, String
value) {
+ ConnectionCountingDataSource connections = new
ConnectionCountingDataSource();
+ TestableMysqlTableOperations ops = new
TestableMysqlTableOperations(connections);
+ JdbcColumn column =
+
JdbcColumn.builder().withName("col1").withType(Types.IntegerType.get()).build();
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ ops.create(
+ "database",
+ "test_table",
+ new JdbcColumn[] {column},
+ "comment",
+ Collections.singletonMap(key, value),
+ Transforms.EMPTY_TRANSFORM,
+ Distributions.NONE,
+ Indexes.EMPTY_INDEXES));
+ Assertions.assertEquals(0, connections.totalBorrows());
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> ops.alterTable("database", "test_table",
TableChange.setProperty(key, value)));
+ Assertions.assertTrue(connections.executedSql().isEmpty());
+ Assertions.assertEquals(0, connections.borrowed());
+ // branch-1.3 generates ALTER SQL inside the execution connection;
validation still prevents
+ // DDL.
+ Assertions.assertEquals(1, connections.peakBorrowed());
+ Assertions.assertEquals(1, connections.totalBorrows());
+ }
+
+ @Test
+ void testUnsupportedSqlOptionNamesSupportedOptions() {
+ TestableMysqlTableOperations ops = new TestableMysqlTableOperations();
+ JdbcColumn column =
+
JdbcColumn.builder().withName("col1").withType(Types.IntegerType.get()).build();
+ IllegalArgumentException error =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ ops.generateCreateTableSql(
+ "test_table",
+ new JdbcColumn[] {column},
+ null,
+ Map.of("ROW_FORMAT", "DYNAMIC"),
+ Transforms.EMPTY_TRANSFORM,
+ Distributions.NONE,
+ Indexes.EMPTY_INDEXES));
+ Assertions.assertTrue(
+ error.getMessage().contains("Supported options: AUTO_INCREMENT,
ENGINE"),
+ error.getMessage());
+ }
+
+ @Test
+ void testSupportedSqlOptionsAreNormalized() {
+ TestableMysqlTableOperations ops = new TestableMysqlTableOperations();
+ JdbcColumn column =
+
JdbcColumn.builder().withName("col1").withType(Types.IntegerType.get()).build();
+ String sql =
+ ops.generateCreateTableSql(
+ "test_table",
+ new JdbcColumn[] {column},
+ null,
+ Map.of("ENGINE", "InNoDB", "AUTO_INCREMENT", "+0042"),
+ Transforms.EMPTY_TRANSFORM,
+ Distributions.NONE,
+ Indexes.EMPTY_INDEXES);
+ Assertions.assertTrue(sql.contains("ENGINE = InnoDB"), sql);
+ Assertions.assertTrue(sql.contains("AUTO_INCREMENT = 42"), sql);
+ }
+
@Test
public void testCreateTableWithEmptyStringDefaultValue() {
TestableMysqlTableOperations ops = new TestableMysqlTableOperations();
@@ -147,4 +229,14 @@ public class TestMysqlTableOperationsSqlGeneration {
Assertions.assertTrue(alterSql.contains("ADD COLUMN `col2`"), alterSql);
Assertions.assertTrue(alterSql.contains("COMMENT 'owner\\\\''s
\"comment\"; --'"), alterSql);
}
+
+ private static Stream<Arguments> invalidTableOptions() {
+ return Stream.of(
+ Arguments.of("ENGINE", "ABC"),
+ Arguments.of("ENGINE", "InnoDB, ROW_FORMAT=DYNAMIC"),
+ Arguments.of("AUTO_INCREMENT", "1; DROP TABLE t"),
+ Arguments.of("ROW_FORMAT", "DYNAMIC"),
+ Arguments.of("ENGINE; DROP TABLE t", "MEMORY"),
+ Arguments.of("AUTO_INCREMENT", null));
+ }
}
diff --git a/docs/jdbc-clickhouse-catalog.md b/docs/jdbc-clickhouse-catalog.md
index c0f7ee27e4..fdc7be7a56 100644
--- a/docs/jdbc-clickhouse-catalog.md
+++ b/docs/jdbc-clickhouse-catalog.md
@@ -56,7 +56,7 @@ When using the JDBC catalog you must provide `jdbc-url`,
`jdbc-driver`, `jdbc-us
| `jdbc-password` | JDBC password
| (none) | Yes |
| `jdbc.pool.min-size` | Minimum pool size
| `2` | No |
| `jdbc.pool.max-size` | Maximum pool size
| `10` | No |
-| `jdbc.pool.max-wait-ms` | Max wait time for a connection
| `30000` | No |
+| `jdbc.pool.max-wait-ms` | The maximum Duration that the pool will wait for a
connection to be returned; takes precedence over
`gravitino.bypass.maxWaitMillis`. | `30000` | No |
### Create a ClickHouse Catalog
diff --git a/docs/jdbc-doris-catalog.md b/docs/jdbc-doris-catalog.md
index 47bf6e02aa..49e576aceb 100644
--- a/docs/jdbc-doris-catalog.md
+++ b/docs/jdbc-doris-catalog.md
@@ -50,7 +50,7 @@ Besides the [common catalog
properties](./gravitino-server-config.md#catalog-pro
| `jdbc.pool.min-size` | The minimum number of connections in the pool. `2`
by default.
| `2` | No |
| `jdbc.pool.max-size` | The maximum number of connections in the pool.
`10` by default.
| `10` | No |
| `replication_num` | The number of replications for the table. If not
specified and the number of backend servers less than 3, then the default value
is 1; If not specified and the number of backend servers greater or equals to
3, the default value (3) in Doris server will be used. For more, see the
[doc](https://doris.apache.org/docs/1.2/sql-manual/sql-reference/Data-Definition-Statements/Create/CREATE-TABLE/)
| `1` or `3` | No |
-| `jdbc.pool.max-wait-ms` | The maximum Duration that the pool will wait for a
connection to be returned. `30000` by default.
| `30000` | No |
+| `jdbc.pool.max-wait-ms` | The maximum Duration that the pool will wait for a
connection to be returned; takes precedence over
`gravitino.bypass.maxWaitMillis`. | `30000` | No |
Before using the Doris Catalog, you must download the corresponding JDBC
driver to the `catalogs/jdbc-doris/libs` directory.
Gravitino doesn't package the JDBC driver for Doris due to licensing issues.
diff --git a/docs/jdbc-mysql-catalog.md b/docs/jdbc-mysql-catalog.md
index bd2964861b..534baa1bad 100644
--- a/docs/jdbc-mysql-catalog.md
+++ b/docs/jdbc-mysql-catalog.md
@@ -49,7 +49,7 @@ Besides the [common catalog
properties](./gravitino-server-config.md#catalog-pro
| `jdbc-password` | The JDBC password.
| (none) | Yes |
| `jdbc.pool.min-size` | The minimum number of connections in the pool. `2`
by default. | `2` | No |
| `jdbc.pool.max-size` | The maximum number of connections in the pool.
`10` by default. | `10` | No
|
-| `jdbc.pool.max-wait-ms` | The maximum Duration that the pool will wait for a
connection to be returned. `30000` by default. | `30000` | No |
+| `jdbc.pool.max-wait-ms` | The maximum Duration that the pool will wait for a
connection to be returned; takes precedence over
`gravitino.bypass.maxWaitMillis`. | `30000` | No |
:::caution
Gravitino does not package the MySQL JDBC driver, because MySQL Connector/J is
@@ -222,7 +222,9 @@ Although MySQL itself does not support table properties,
Gravitino offers table
:::
:::caution
-- Doesn't support remove table properties. You can only add or modify
properties, not delete properties.
+For MySQL table options, use the Gravitino property names listed below.
Unknown properties, including SQL option names such as `ENGINE` or
`ROW_FORMAT`, are rejected. Use the Gravitino property name `engine` instead.
+
+These properties can only be specified when creating a table; they cannot be
changed or removed afterwards. Values are validated against their declared
types before DDL is executed.
:::
| Property Name | Description
| Default Value | Required | Reserved | Immutable |
diff --git a/docs/jdbc-oceanbase-catalog.md b/docs/jdbc-oceanbase-catalog.md
index 9e0b99e7a5..40c99f58a8 100644
--- a/docs/jdbc-oceanbase-catalog.md
+++ b/docs/jdbc-oceanbase-catalog.md
@@ -47,7 +47,7 @@ Besides the [common catalog
properties](./gravitino-server-config.md#catalog-pro
| `jdbc-password` | The JDBC password.
| (none) | Yes |
| `jdbc.pool.min-size` | The minimum number of connections in the pool. `2`
by default.
| `2` | No |
| `jdbc.pool.max-size` | The maximum number of connections in the pool.
`10` by default.
| `10` | No |
-| `jdbc.pool.max-wait-ms` | The maximum Duration that the pool will wait for a
connection to be returned. `30000` by default.
| `30000` | No |
+| `jdbc.pool.max-wait-ms` | The maximum Duration that the pool will wait for a
connection to be returned; takes precedence over
`gravitino.bypass.maxWaitMillis`. | `30000` | No |
:::caution
Gravitino does not package the OceanBase JDBC driver due to licensing, so you
diff --git a/docs/jdbc-postgresql-catalog.md b/docs/jdbc-postgresql-catalog.md
index 70ab6b04bc..b5def139a8 100644
--- a/docs/jdbc-postgresql-catalog.md
+++ b/docs/jdbc-postgresql-catalog.md
@@ -48,7 +48,7 @@ Besides the [common catalog
properties](./gravitino-server-config.md#catalog-pro
| `jdbc-password` | The JDBC password.
| (none) | Yes |
| `jdbc.pool.min-size` | The minimum number of connections in the pool. `2`
by default.
| `2` | No |
| `jdbc.pool.max-size` | The maximum number of connections in the pool.
`10` by default.
| `10` | No |
-| `jdbc.pool.max-wait-ms` | The maximum Duration that the pool will wait for a
connection to be returned. `30000` by default.
| `30000` | No |
+| `jdbc.pool.max-wait-ms` | The maximum Duration that the pool will wait for a
connection to be returned; takes precedence over
`gravitino.bypass.maxWaitMillis`. | `30000` | No |
:::caution
Download the corresponding JDBC driver to the `catalogs/jdbc-postgresql/libs`
directory.
diff --git a/docs/jdbc-starrocks-catalog.md b/docs/jdbc-starrocks-catalog.md
index a03d268cbe..ad2d9850ce 100644
--- a/docs/jdbc-starrocks-catalog.md
+++ b/docs/jdbc-starrocks-catalog.md
@@ -48,7 +48,7 @@ Besides the [common catalog
properties](./gravitino-server-config.md#catalog-pro
| `jdbc-password` | The JDBC password.
| (none) | Yes |
| `jdbc.pool.min-size` | The minimum number of connections in the pool. `2`
by default. | `2` | No |
| `jdbc.pool.max-size` | The maximum number of connections in the pool.
`10` by default. | `10` | No |
-| `jdbc.pool.max-wait-ms` | The maximum Duration that the pool will wait for a
connection to be returned. `30000` by default. | `30000` | No |
+| `jdbc.pool.max-wait-ms` | The maximum Duration that the pool will wait for a
connection to be returned; takes precedence over
`gravitino.bypass.maxWaitMillis`. | `30000` | No |
Before using the StarRocks Catalog, you must download the corresponding JDBC
driver to the `catalogs/jdbc-starrocks/libs` directory.
Gravitino doesn't package the JDBC driver for StarRocks due to licensing
issues.
diff --git
a/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/jdbc/JdbcPropertiesConverter.java
b/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/jdbc/JdbcPropertiesConverter.java
index 611952e02d..d0994d9259 100644
---
a/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/jdbc/JdbcPropertiesConverter.java
+++
b/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/jdbc/JdbcPropertiesConverter.java
@@ -19,8 +19,10 @@
package org.apache.gravitino.flink.connector.jdbc;
+import com.google.common.collect.ImmutableSet;
import java.util.HashMap;
import java.util.Map;
+import java.util.Set;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.apache.commons.lang3.StringUtils;
@@ -35,6 +37,13 @@ public abstract class JdbcPropertiesConverter
private static final Pattern jdbcUrlPattern =
Pattern.compile("(jdbc:\\w+://[^:/]+(?::\\d+)?)");
+ private static final Set<String> INJECTED_TABLE_OPTIONS =
+ ImmutableSet.of(
+ JdbcPropertiesConstants.FLINK_JDBC_TABLE_DATABASE_URL,
+ JdbcPropertiesConstants.FLINK_JDBC_TABLE_NAME,
+ JdbcPropertiesConstants.FLINK_JDBC_USER,
+ JdbcPropertiesConstants.FLINK_JDBC_PASSWORD);
+
@Override
public Map<String, String> toGravitinoCatalogProperties(Configuration
flinkConf) {
Map<String, String> gravitinoCatalogProperties =
@@ -86,6 +95,18 @@ public abstract class JdbcPropertiesConverter
return JdbcPropertiesConstants.gravitinoToFlinkMap.get(configKey);
}
+ /**
+ * Drops the connection options that {@link #toFlinkTableProperties} injects
for scans. Flink
+ * copies them into the options of {@code CREATE TABLE ... LIKE}, but they
describe the catalog
+ * connection rather than the table, and must not reach the server as table
properties.
+ */
+ @Override
+ public Map<String, String> toGravitinoTableProperties(Map<String, String>
flinkProperties) {
+ Map<String, String> gravitinoProperties = new HashMap<>(flinkProperties);
+ gravitinoProperties.keySet().removeAll(INJECTED_TABLE_OPTIONS);
+ return gravitinoProperties;
+ }
+
@Override
public Map<String, String> toFlinkTableProperties(
Map<String, String> flinkCatalogProperties,
diff --git
a/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/FlinkCommonIT.java
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/FlinkCommonIT.java
index 159fe00775..222a5a370c 100644
---
a/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/FlinkCommonIT.java
+++
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/FlinkCommonIT.java
@@ -95,6 +95,15 @@ public abstract class FlinkCommonIT extends FlinkEnvIT {
return true;
}
+ /**
+ * Whether the catalog accepts arbitrary table properties, such as {@code
'test key' = 'test
+ * value'}, at creation time. Catalogs that validate table properties
against a fixed set reject
+ * them.
+ */
+ protected boolean supportsCustomTableProperties() {
+ return true;
+ }
+
protected boolean defaultValueWithNullLiterals() {
return false;
}
@@ -296,16 +305,15 @@ public abstract class FlinkCommonIT extends FlinkEnvIT {
"CREATE TABLE %s "
+ "(string_type STRING COMMENT 'string_type', "
+ " double_type DOUBLE COMMENT 'double_type')"
- + " COMMENT '%s' WITH ("
- + "'%s' = '%s')",
- tableName, comment, key, value);
+ + " COMMENT '%s'%s",
+ tableName, comment, customTablePropertiesClause(key, value));
TestUtils.assertTableResult(result, ResultKind.SUCCESS);
Table table =
catalog.asTableCatalog().loadTable(NameIdentifier.of(databaseName, tableName));
Assertions.assertNotNull(table);
Assertions.assertEquals(comment, table.comment());
- if (supportTablePropertiesOperation()) {
+ if (supportTablePropertiesOperation() &&
supportsCustomTableProperties()) {
Assertions.assertEquals(value, table.properties().get(key));
}
Column[] columns =
@@ -350,9 +358,8 @@ public abstract class FlinkCommonIT extends FlinkEnvIT {
+ " cc int,"
+ " PRIMARY KEY (aa,bb) NOT ENFORCED"
+ ")"
- + " COMMENT '%s' WITH ("
- + "'%s' = '%s')",
- tableName, comment, key, value);
+ + " COMMENT '%s'%s",
+ tableName, comment, customTablePropertiesClause(key, value));
Table table =
catalog.asTableCatalog().loadTable(NameIdentifier.of(databaseName, tableName));
Assertions.assertEquals(1, table.index().length);
@@ -492,7 +499,9 @@ public abstract class FlinkCommonIT extends FlinkEnvIT {
NameIdentifier.of(databaseName, "test_desc_table"),
columns,
comment,
- ImmutableMap.of("k1", "v1"));
+ supportsCustomTableProperties()
+ ? ImmutableMap.of("k1", "v1")
+ : ImmutableMap.of());
Optional<org.apache.flink.table.catalog.Catalog> flinkCatalog =
tableEnv.getCatalog(catalog.name());
@@ -708,10 +717,8 @@ public abstract class FlinkCommonIT extends FlinkEnvIT {
"CREATE TABLE %s "
+ "(user_id BIGINT COMMENT 'USER_ID', "
+ " order_amount INT COMMENT 'ORDER_AMOUNT')"
- + " COMMENT 'test comment'"
- + " WITH ("
- + "'%s' = '%s')",
- tableName, "test key", "test value");
+ + " COMMENT 'test comment'%s",
+ tableName, customTablePropertiesClause("test key", "test
value"));
TestUtils.assertTableResult(result, ResultKind.SUCCESS);
result = sql("ALTER TABLE %s MODIFY order_amount INT COMMENT 'new
comment2'", tableName);
TestUtils.assertTableResult(result, ResultKind.SUCCESS);
@@ -1111,4 +1118,8 @@ public abstract class FlinkCommonIT extends FlinkEnvIT {
true,
supportDropCascade());
}
+
+ private String customTablePropertiesClause(String key, String value) {
+ return supportsCustomTableProperties() ? String.format(" WITH ('%s' =
'%s')", key, value) : "";
+ }
}
diff --git
a/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/jdbc/FlinkJdbcMysqlCatalogIT.java
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/jdbc/FlinkJdbcMysqlCatalogIT.java
index 23989bc4cd..1ca152cfc3 100644
---
a/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/jdbc/FlinkJdbcMysqlCatalogIT.java
+++
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/integration/test/jdbc/FlinkJdbcMysqlCatalogIT.java
@@ -26,6 +26,7 @@ import com.google.common.collect.ImmutableMap;
import java.util.Map;
import org.apache.flink.table.api.ValidationException;
import org.apache.gravitino.Catalog;
+import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.credential.CredentialConstants;
import org.apache.gravitino.credential.JdbcCredential;
import org.apache.gravitino.flink.connector.integration.test.FlinkCommonIT;
@@ -57,6 +58,11 @@ public abstract class FlinkJdbcMysqlCatalogIT extends
FlinkCommonIT {
return false;
}
+ @Override
+ protected boolean supportsCustomTableProperties() {
+ return false;
+ }
+
@Override
protected String defaultDatabaseName() {
return MYSQL_CATALOG_MYSQL_IT.name();
@@ -205,4 +211,30 @@ public abstract class FlinkJdbcMysqlCatalogIT extends
FlinkCommonIT {
catalogName, mysqlUrl, mysqlUsername,
mysqlDefaultDatabase)),
"Missing required options are:\n" + "\n" + "password");
}
+
+ @Test
+ public void testCreateTableLikeDoesNotForwardConnectorOptions() {
+ String databaseName = "test_create_table_like_db";
+ doWithSchema(
+ currentCatalog(),
+ databaseName,
+ catalog -> {
+ sql("CREATE TABLE source_table (id INT COMMENT 'id', name STRING
COMMENT 'name')");
+ // LIKE copies the options that getTable() exposes for the source
table, which include the
+ // JDBC connection options the connector injects for scans.
+ sql("CREATE TABLE like_table LIKE source_table");
+
+ Map<String, String> properties =
+ catalog
+ .asTableCatalog()
+ .loadTable(NameIdentifier.of(databaseName, "like_table"))
+ .properties();
+ Assertions.assertFalse(
+
properties.containsKey(JdbcPropertiesConstants.FLINK_JDBC_PASSWORD));
+ Assertions.assertFalse(
+
properties.containsKey(JdbcPropertiesConstants.FLINK_JDBC_TABLE_DATABASE_URL));
+ },
+ true,
+ supportDropCascade());
+ }
}
diff --git
a/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/jdbc/AbstractJdbcPropertiesConverterTestSuite.java
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/jdbc/AbstractJdbcPropertiesConverterTestSuite.java
index d89fbf5401..c40eb6277e 100644
---
a/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/jdbc/AbstractJdbcPropertiesConverterTestSuite.java
+++
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/jdbc/AbstractJdbcPropertiesConverterTestSuite.java
@@ -23,6 +23,7 @@ import com.google.common.collect.ImmutableMap;
import java.util.HashMap;
import java.util.Map;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.catalog.ObjectPath;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -168,4 +169,24 @@ public abstract class
AbstractJdbcPropertiesConverterTestSuite {
Assertions.assertTrue(ex.getMessage().contains(JdbcPropertiesConstants.GRAVITINO_JDBC_URL));
}
+
+ @Test
+ public void testInjectedTableOptionsDoNotBecomeGravitinoTableProperties() {
+ JdbcPropertiesConverter converter = getConverter(catalogProperties);
+ Map<String, String> flinkTableOptions =
+ new HashMap<>(
+ converter.toFlinkTableProperties(
+ converter.toFlinkCatalogProperties(catalogProperties),
+ ImmutableMap.of(),
+ new ObjectPath("db", "source_table")));
+ // CREATE TABLE ... LIKE copies the scan options alongside the user's own
options.
+ flinkTableOptions.put("engine", "InnoDB");
+
+ Map<String, String> gravitinoProperties =
+ converter.toGravitinoTableProperties(flinkTableOptions);
+
+ Assertions.assertEquals(ImmutableMap.of("engine", "InnoDB"),
gravitinoProperties);
+ Assertions.assertEquals(
+ password,
flinkTableOptions.get(JdbcPropertiesConstants.FLINK_JDBC_PASSWORD));
+ }
}
diff --git
a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/JdbcPropertiesConverter.java
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/JdbcPropertiesConverter.java
index 7bd2a86b85..7c80232d5e 100644
---
a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/JdbcPropertiesConverter.java
+++
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/JdbcPropertiesConverter.java
@@ -24,6 +24,7 @@ import com.google.common.collect.ImmutableMap;
import java.util.HashMap;
import java.util.Map;
import org.apache.gravitino.spark.connector.PropertiesConverter;
+import org.apache.spark.sql.connector.catalog.TableCatalog;
public class JdbcPropertiesConverter implements PropertiesConverter {
@@ -84,7 +85,10 @@ public class JdbcPropertiesConverter implements
PropertiesConverter {
}
return new HashMap<>();
} else {
- return new HashMap<>(properties);
+ Map<String, String> tableProperties = new HashMap<>(properties);
+ // Spark adds ownership to CREATE TABLE properties; JDBC ownership is
not a table option.
+
tableProperties.keySet().removeIf(TableCatalog.PROP_OWNER::equalsIgnoreCase);
+ return tableProperties;
}
}
diff --git
a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/jdbc/TestJdbcPropertiesConverter.java
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/jdbc/TestJdbcPropertiesConverter.java
index 5d3e4d065f..1703e355d0 100644
---
a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/jdbc/TestJdbcPropertiesConverter.java
+++
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/jdbc/TestJdbcPropertiesConverter.java
@@ -23,11 +23,53 @@ import com.google.common.collect.ImmutableMap;
import java.util.Map;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
public class TestJdbcPropertiesConverter {
private final JdbcPropertiesConverter jdbcPropertiesConverter =
JdbcPropertiesConverter.getInstance();
+ @Test
+ void testGetPGInstance() {
+ JdbcPropertiesConverter instance = JdbcPropertiesConverter.getPGInstance();
+ Assertions.assertNotNull(instance);
+ Assertions.assertSame(instance, JdbcPropertiesConverter.getPGInstance());
+ Assertions.assertNotSame(instance, JdbcPropertiesConverter.getInstance());
+
+ Map<String, String> emptyProps =
instance.toGravitinoTableProperties(ImmutableMap.of());
+ Assertions.assertTrue(emptyProps.isEmpty());
+
+ Map<String, String> ownerProps =
+ instance.toGravitinoTableProperties(ImmutableMap.of("owner", "alice"));
+ Assertions.assertTrue(ownerProps.isEmpty());
+
+ Map<String, String> upperOwnerProps =
+ instance.toGravitinoTableProperties(ImmutableMap.of("OWNER", "bob"));
+ Assertions.assertTrue(upperOwnerProps.isEmpty());
+
+ Assertions.assertThrows(
+ UnsupportedOperationException.class,
+ () -> instance.toGravitinoTableProperties(ImmutableMap.of("someKey",
"value")));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"owner", "OWNER"})
+ void testSparkOwnershipIsNotForwardedAsATableOption(String ownerKey) {
+ Map<String, String> properties =
+ ImmutableMap.of(ownerKey, "alice", "engine", "InnoDB", "unknown-key",
"value");
+
+ Map<String, String> converted =
jdbcPropertiesConverter.toGravitinoTableProperties(properties);
+
+ Assertions.assertEquals(ImmutableMap.of("engine", "InnoDB", "unknown-key",
"value"), converted);
+ // Keep caller-owned input intact and forward unknown user keys for
server-side validation.
+ Assertions.assertEquals("alice", properties.get(ownerKey));
+ Assertions.assertTrue(
+ jdbcPropertiesConverter
+ .toGravitinoTableProperties(ImmutableMap.of(ownerKey, "alice"))
+ .isEmpty());
+ }
+
@Test
void testCatalogProperties() {
String url = "jdbc-url";
