This is an automated email from the ASF dual-hosted git repository.
yuqi1129 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new 3ce91977a2 [#13945] fix(jdbc): Honor pool wait settings and validate
MySQL table properties (#13946)
3ce91977a2 is described below
commit 3ce91977a2d137af39321716a9fb5813922afceb
Author: Qi Yu <[email protected]>
AuthorDate: Tue Oct 20 22:12:13 2026 +0800
[#13945] fix(jdbc): Honor pool wait settings and validate MySQL table
properties (#13946)
### What changes were proposed in this pull request?
- Register and forward `jdbc.pool.max-wait-ms` so it reaches the JDBC
connection pool. As with `jdbc.pool.max-idle`, an explicit canonical
value takes precedence over `gravitino.bypass.maxWaitMillis`; the
canonical default no longer overrides a bypassed `maxWaitMillis`.
- Reject undeclared MySQL table properties instead of silently dropping
them, while retaining internal identifier and credential metadata
handling.
- Decode supported MySQL options using their declared enum/integer types
and render canonical SQL values. Apply the same validation to
create/alter SQL generation.
- Remove Spark's implicit `owner` metadata in the JDBC connector before
forwarding table options, preserving normal Spark table creation under
strict MySQL validation.
- Strip the JDBC scan options the Flink connector injects (`url`,
`table-name`, `username`, `password`) before forwarding table options,
so `CREATE TABLE ... LIKE` keeps working and the catalog password is no
longer sent as a table property. Same-named options written explicitly
in `WITH` are dropped too; they never took effect, because scan options
are always regenerated from the catalog connection.
- Make the shared Flink catalog ITs send arbitrary `WITH` options only
to catalogs that accept custom table properties; the MySQL Flink IT opts
out.
- Clarify the MySQL property names and immutable-property policy in the
catalog documentation, and list the supported keys in the rejection
message.
- Document `jdbc.pool.max-wait-ms` and its bypass precedence in every
JDBC catalog page (adds the missing Hologres row).
### Why are the changes needed?
The canonical pool-wait setting currently falls back to `30000` ms.
Unknown MySQL table properties are accepted and then discarded by
mapping on both create and alter. This patch corrects these two
property-handling paths; it does not claim that unknown properties
bypass immutability through the public API.
Fix: #13945
### Does this PR introduce _any_ user-facing change?
Configured pool wait times take effect, whether set through
`jdbc.pool.max-wait-ms` or `gravitino.bypass.maxWaitMillis`. Values that
were previously ignored now apply after upgrade: a stored non-positive
`jdbc.pool.max-wait-ms` fails catalog initialization, and a bypassed
`maxWaitMillis=-1` means an unbounded wait. MySQL create/alter requests
with unknown table properties now fail with an error naming the key.
Supported values are validated before DDL. Shared metadata keys that
every entity accepts (the internal identifier and credential-vending
keys) are still accepted and not turned into MySQL options, since they
are not table options. `engine` and `auto-increment-offset` remain
immutable, and strict unknown-key validation is scoped to MySQL. Spark
JDBC no longer forwards its reserved `owner` metadata as a table option;
other user-supplied keys continue to reach the server for
catalog-specific handling.
### How was this patch tested?
- Full JDBC common and MySQL non-Docker unit suites: 106 and 29 tests
passed.
- Regressions exercise catalog initialization through the actual pool,
catalog create/alter property mapping, internal metadata handling, and
create/alter SQL generation before any connection is borrowed.
- Separate mutation checks revert the pool registration, MySQL metadata
validation, and SQL validation; the corresponding regression cases fail
as expected.
- MySQL catalog module with Docker enabled: 51 tests passed, including
real create/alter property round trips.
- Full Spark 3.5 and 4.0 non-Docker suites: 233 and 229 tests passed,
including implicit-owner removal and preservation of unknown user keys
for server-side validation.
- Reverting Spark owner removal makes both owner regression cases fail;
restoring the fix passes all four converter tests.
- Pool wait precedence covers canonical, prefixed canonical and DBCP
`maxWaitMillis` bypass combinations (6 cases); reverting the precedence
guard fails the bypass-only case.
- Flink converter suites (MySQL, PostgreSQL) cover stripping injected
scan options; removing the stripping fails both. A new `CREATE TABLE ...
LIKE` case in `FlinkJdbcMysqlCatalogIT119` fails without the fix
(`Unsupported MySQL table property: password`) and passes with it.
- `FlinkJdbcMysqlCatalogIT119` with Docker MySQL: 30 tests, 0 failures,
including the LIKE case and the four create-table cases that previously
sent unsupported options.
- JDBC common, MySQL, and Spark common, and Flink common Spotless
checks.
---
.../jdbc/JdbcCatalogPropertiesMetadata.java | 7 +
.../catalog/jdbc/utils/DataSourceUtils.java | 7 +-
.../jdbc/TestJdbcCatalogPropertiesMetadata.java | 86 ++++++++
.../mysql/MysqlTablePropertiesMetadata.java | 34 +++-
.../mysql/operation/MysqlTableOperations.java | 22 +-
.../mysql/TestMysqlTablePropertiesMetadata.java | 222 +++++++++++++++++++++
.../mysql/operation/TestMysqlTableOperations.java | 11 +-
.../TestMysqlTableOperationsSqlGeneration.java | 88 ++++++++
docs/jdbc-clickhouse-catalog.md | 2 +-
docs/jdbc-doris-catalog.md | 2 +-
docs/jdbc-hologres-catalog.md | 21 +-
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 | 19 ++
21 files changed, 598 insertions(+), 50 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 3f933b0975..be57c26f3e 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
@@ -47,6 +47,7 @@ public class JdbcCatalogPropertiesMetadata extends
BaseCatalogPropertiesMetadata
JdbcConfig.POOL_MIN_SIZE.getKey(),
JdbcConfig.POOL_MAX_SIZE.getKey(),
JdbcConfig.POOL_MAX_IDLE.getKey(),
+ JdbcConfig.POOL_MAX_WAIT_MS.getKey(),
JdbcConfig.TEST_ON_BORROW.getKey(),
JdbcConfig.CONNECT_TIMEOUT_MS.getKey(),
JdbcConfig.VERIFY_ON_CREATE.getKey());
@@ -116,6 +117,12 @@ public class JdbcCatalogPropertiesMetadata extends
BaseCatalogPropertiesMetadata
JdbcConfig.POOL_MAX_IDLE.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 c2d224b7a1..b2f61e19ff 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
@@ -166,7 +166,12 @@ public class DataSourceUtils {
: jdbcConfig.getPoolMinSize());
// Validate connections on borrow when enabled.
basicDataSource.setTestOnBorrow(jdbcConfig.getTestOnBorrow());
- basicDataSource.setMaxWait(Duration.ofMillis(jdbcConfig.getMaxWaitMs()));
+ // Same precedence as maxIdle: 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/socket timeout as DBCP connection
properties (not pool keys,
// so no collision with UNSAFE_POOL_PROPERTIES and the pool stays lazy).
These bound a single
// connection and take effect on the first real connection — i.e.
create-time verification — so
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 b6fd8a56bf..60a4e95a34 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,7 +20,10 @@ 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;
@@ -36,12 +39,14 @@ import
org.apache.gravitino.catalog.jdbc.operation.SqliteTableOperations;
import
org.apache.gravitino.catalog.jdbc.utils.TestDataSourceValidation.UnsupportedValidationDriver;
import org.apache.gravitino.connector.CatalogInfo;
import org.apache.gravitino.exceptions.ConnectionFailedException;
+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 {
@@ -68,6 +73,87 @@ 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());
+ }
+ }
+
+ @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..1b73a3f704 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 and credential settings, 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 7f80ceba9c..af5b4becc4 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
@@ -49,6 +49,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;
@@ -65,6 +66,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 =
@@ -134,7 +137,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", "")));
}
@@ -578,11 +581,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..16893e6b6b
--- /dev/null
+++
b/catalogs/catalog-jdbc-mysql/src/test/java/org/apache/gravitino/catalog/mysql/TestMysqlTablePropertiesMetadata.java
@@ -0,0 +1,222 @@
+/*
+ * 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",
+ "credential-providers",
+ "test-provider"));
+ 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 9a871df0e3..e15d53b0a3 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
@@ -39,7 +39,6 @@ import org.apache.gravitino.catalog.jdbc.JdbcTable;
import org.apache.gravitino.catalog.jdbc.utils.DataSourceUtils;
import
org.apache.gravitino.catalog.mysql.converter.MysqlColumnDefaultValueConverter;
import org.apache.gravitino.catalog.mysql.converter.MysqlTypeConverter;
-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;
@@ -157,15 +156,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 7f240760a0..a8292a95cb 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.Column;
import org.apache.gravitino.rel.TableChange;
@@ -34,6 +37,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 {
@@ -44,6 +50,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");
}
@@ -70,6 +81,73 @@ 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.assertThrows(
+ IllegalArgumentException.class,
+ () -> ops.alterTable("database", "test_table",
TableChange.setProperty(key, value)));
+ Assertions.assertTrue(connections.executedSql().isEmpty());
+ Assertions.assertEquals(0, connections.borrowed());
+ Assertions.assertEquals(0, connections.peakBorrowed());
+ }
+
+ @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();
@@ -210,4 +288,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 370ba3758a..4b6e64db22 100644
--- a/docs/jdbc-clickhouse-catalog.md
+++ b/docs/jdbc-clickhouse-catalog.md
@@ -62,7 +62,7 @@ When using the JDBC catalog you must provide `jdbc-url`,
`jdbc-driver`, `jdbc-us
| `jdbc.pool.min-size` | Minimum pool size
| `2` | No
|
| `jdbc.pool.max-size` | Maximum pool size
| `10` | No
|
| `jdbc.pool.max-idle` | Maximum idle connections retained per catalog per
server; capped by `jdbc.pool.max-size`; takes precedence over
`gravitino.bypass.maxIdle`. Idle connections are not evicted by default. | `8`
| 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 0dea47a4f8..15090280c1 100644
--- a/docs/jdbc-doris-catalog.md
+++ b/docs/jdbc-doris-catalog.md
@@ -53,7 +53,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
|
| `jdbc.pool.max-idle` | Maximum idle connections retained per catalog per
server; capped by `jdbc.pool.max-size`; takes precedence over
`gravitino.bypass.maxIdle`. Idle connections are not evicted by default. | `8`
| 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-hologres-catalog.md b/docs/jdbc-hologres-catalog.md
index 4bd30f1fa4..e2844b88cb 100644
--- a/docs/jdbc-hologres-catalog.md
+++ b/docs/jdbc-hologres-catalog.md
@@ -46,16 +46,17 @@ Check the relevant data source configuration in [data
source properties](https:/
If you use a JDBC catalog, you must provide `jdbc-url`, `jdbc-driver`,
`jdbc-database`, `jdbc-user` and `jdbc-password` to catalog properties.
Besides the [common catalog
properties](./gravitino-server-config.md#catalog-properties-configuration), the
Hologres catalog has the following properties:
-| Configuration item | Description
| Default value | Required |
-|----------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|---------------|----------|
-| `jdbc-url` | JDBC URL for connecting to the database. For example,
`jdbc:postgresql://hgprecn-cn-xxx.hologres.aliyuncs.com:80/my_database`
| (none) | Yes |
-| `jdbc-driver` | The driver of the JDBC connection. Must be
`org.postgresql.Driver`.
| (none) |
Yes |
-| `jdbc-database` | The database name. This is mandatory for Hologres.
| (none) | Yes |
-| `jdbc-user` | The JDBC user name (AccessKey ID or database
username).
| (none) |
Yes |
-| `jdbc-password` | The JDBC password (AccessKey Secret or database
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-idle` | Maximum idle connections retained per catalog per
server; capped by `jdbc.pool.max-size`; takes precedence over
`gravitino.bypass.maxIdle`. Idle connections are not evicted by default. | `8`
| No |
+| Configuration item | Description
| Default value |
Required |
+|-------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|---------------|----------|
+| `jdbc-url` | JDBC URL for connecting to the database. For
example,
`jdbc:postgresql://hgprecn-cn-xxx.hologres.aliyuncs.com:80/my_database`
| (none) | Yes |
+| `jdbc-driver` | The driver of the JDBC connection. Must be
`org.postgresql.Driver`.
| (none) |
Yes |
+| `jdbc-database` | The database name. This is mandatory for Hologres.
| (none) | Yes
|
+| `jdbc-user` | The JDBC user name (AccessKey ID or database
username).
| (none) |
Yes |
+| `jdbc-password` | The JDBC password (AccessKey Secret or database
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-idle` | Maximum idle connections retained per catalog per
server; capped by `jdbc.pool.max-size`; takes precedence over
`gravitino.bypass.maxIdle`. Idle connections are not evicted by default. | `8`
| 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
Hologres uses the PostgreSQL JDBC Driver (version 42.3.2 or later
recommended). You need to download the PostgreSQL JDBC Driver and place it in
the `catalogs/jdbc-hologres/libs` directory under the Gravitino distribution
(e.g., `distribution/package/catalogs/jdbc-hologres/libs` or
`distribution/package-all/catalogs/jdbc-hologres/libs`).
diff --git a/docs/jdbc-mysql-catalog.md b/docs/jdbc-mysql-catalog.md
index ec15f1736e..14bacae6bc 100644
--- a/docs/jdbc-mysql-catalog.md
+++ b/docs/jdbc-mysql-catalog.md
@@ -53,7 +53,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
|
| `jdbc.pool.max-idle` | Maximum idle connections retained per catalog per
server; capped by `jdbc.pool.max-size`; takes precedence over
`gravitino.bypass.maxIdle`. Idle connections are not evicted by default. | `8`
| 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
@@ -214,7 +214,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.
Metadata keys shared by every entity, such as the credential-vending settings,
are accepted but not applied to MySQL tables.
+
+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 192af85017..861597d83e 100644
--- a/docs/jdbc-oceanbase-catalog.md
+++ b/docs/jdbc-oceanbase-catalog.md
@@ -51,7 +51,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
|
| `jdbc.pool.max-idle` | Maximum idle connections retained per catalog per
server; capped by `jdbc.pool.max-size`; takes precedence over
`gravitino.bypass.maxIdle`. Idle connections are not evicted by default. | `8`
| 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 eac6d40ee0..3bc301809f 100644
--- a/docs/jdbc-postgresql-catalog.md
+++ b/docs/jdbc-postgresql-catalog.md
@@ -52,7 +52,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
|
| `jdbc.pool.max-idle` | Maximum idle connections retained per catalog per
server; capped by `jdbc.pool.max-size`; takes precedence over
`gravitino.bypass.maxIdle`. Idle connections are not evicted by default. | `8`
| 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 cb87bad97e..318ea46114 100644
--- a/docs/jdbc-starrocks-catalog.md
+++ b/docs/jdbc-starrocks-catalog.md
@@ -52,7 +52,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
|
| `jdbc.pool.max-idle` | Maximum idle connections retained per catalog per
server; capped by `jdbc.pool.max-size`; takes precedence over
`gravitino.bypass.maxIdle`. Idle connections are not evicted by default. | `8`
| 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 3f802bda82..16dbe33cc0 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
@@ -112,6 +112,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;
}
@@ -315,16 +324,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 =
@@ -369,9 +377,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);
@@ -511,7 +518,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());
@@ -730,10 +739,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);
@@ -1133,4 +1140,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 4c08854cf5..77f0a4ea4b 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();
@@ -208,4 +214,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 cbcac2fce6..f4361d4e52 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 {
@@ -86,7 +87,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 33b704f11f..7de04a6fda 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,6 +23,8 @@ 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 =
@@ -51,6 +53,23 @@ public class TestJdbcPropertiesConverter {
() -> 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";
