This is an automated email from the ASF dual-hosted git repository. github-actions[bot] pushed a commit to branch cherry-pick-3ce91977-to-branch-1.3 in repository https://gitbox.apache.org/repos/asf/gravitino.git
commit ed41da4e743862387b04f7711af1407d8a1b8acc 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. # Conflicts: # catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/JdbcCatalogPropertiesMetadata.java # catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/utils/DataSourceUtils.java # catalogs/catalog-jdbc-common/src/test/java/org/apache/gravitino/catalog/jdbc/TestJdbcCatalogPropertiesMetadata.java # catalogs/catalog-jdbc-mysql/src/test/java/org/apache/gravitino/catalog/mysql/operation/TestMysqlTableOperations.java # docs/jdbc-clickhouse-catalog.md # docs/jdbc-doris-catalog.md # docs/jdbc-hologres-catalog.md # docs/jdbc-mysql-catalog.md # docs/jdbc-oceanbase-catalog.md # docs/jdbc-postgresql-catalog.md # docs/jdbc-starrocks-catalog.md # spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/jdbc/TestJdbcPropertiesConverter.java --- .../jdbc/JdbcCatalogPropertiesMetadata.java | 22 ++ .../catalog/jdbc/utils/DataSourceUtils.java | 13 ++ .../jdbc/TestJdbcCatalogPropertiesMetadata.java | 97 +++++++++ .../mysql/MysqlTablePropertiesMetadata.java | 34 +++- .../mysql/operation/MysqlTableOperations.java | 22 +- .../mysql/TestMysqlTablePropertiesMetadata.java | 222 +++++++++++++++++++++ .../mysql/operation/TestMysqlTableOperations.java | 15 +- .../TestMysqlTableOperationsSqlGeneration.java | 88 ++++++++ docs/jdbc-clickhouse-catalog.md | 13 ++ docs/jdbc-doris-catalog.md | 13 ++ docs/jdbc-hologres-catalog.md | 14 ++ docs/jdbc-mysql-catalog.md | 17 +- docs/jdbc-oceanbase-catalog.md | 13 ++ docs/jdbc-postgresql-catalog.md | 14 ++ docs/jdbc-starrocks-catalog.md | 13 ++ .../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 | 45 +++++ 21 files changed, 738 insertions(+), 32 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..976fb04a58 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,11 @@ public class JdbcCatalogPropertiesMetadata extends BaseCatalogPropertiesMetadata JdbcConfig.PASSWORD.getKey(), JdbcConfig.POOL_MIN_SIZE.getKey(), JdbcConfig.POOL_MAX_SIZE.getKey(), +<<<<<<< HEAD +======= + JdbcConfig.POOL_MAX_IDLE.getKey(), + JdbcConfig.POOL_MAX_WAIT_MS.getKey(), +>>>>>>> 3ce91977a ([#13945] fix(jdbc): Honor pool wait settings and validate MySQL table properties (#13946)) JdbcConfig.TEST_ON_BORROW.getKey(), JdbcConfig.CONNECT_TIMEOUT_MS.getKey(), JdbcConfig.VERIFY_ON_CREATE.getKey()); @@ -107,6 +112,23 @@ public class JdbcCatalogPropertiesMetadata extends BaseCatalogPropertiesMetadata JdbcConfig.POOL_MAX_SIZE.getDefaultValue(), false /* hidden */, false /* reserved */), +<<<<<<< HEAD +======= + integerPropertyEntry( + JdbcConfig.POOL_MAX_IDLE.getKey(), + JdbcConfig.POOL_MAX_IDLE.getDoc(), + false /* required */, + false /* immutable */, + 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 */), +>>>>>>> 3ce91977a ([#13945] fix(jdbc): Honor pool wait settings and validate MySQL table properties (#13946)) 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..50fc33f016 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 @@ -121,12 +121,25 @@ public class DataSourceUtils { // Set each time a connection is taken out from the connection pool, a test statement will be // executed to confirm whether the connection is valid. basicDataSource.setTestOnBorrow(jdbcConfig.getTestOnBorrow()); +<<<<<<< HEAD basicDataSource.setValidationQuery(POOL_TEST_QUERY); 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. // Operator-set values are preserved. +======= + // 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 + // an unreachable backend fails fast instead of hanging. Operator-set values are preserved. +>>>>>>> 3ce91977a ([#13945] fix(jdbc): Honor pool wait settings and validate MySQL table properties (#13946)) JdbcConnectionTimeouts.connectionProperties(jdbcConfig, jdbcUrl) .forEach(basicDataSource::addConnectionProperty); return basicDataSource; 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..66ee661937 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; @@ -32,10 +35,23 @@ 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; +<<<<<<< HEAD +======= +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; +>>>>>>> 3ce91977a ([#13945] fix(jdbc): Honor pool wait settings and validate MySQL table properties (#13946)) import org.apache.gravitino.meta.AuditInfo; import org.apache.gravitino.meta.CatalogEntity; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +<<<<<<< HEAD +======= +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; +import org.junit.jupiter.params.provider.ValueSource; +>>>>>>> 3ce91977a ([#13945] fix(jdbc): Honor pool wait settings and validate MySQL table properties (#13946)) public class TestJdbcCatalogPropertiesMetadata { @@ -61,6 +77,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 813f44b6de..625c917e44 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..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 f4cd86e9e6..e4282bdebe 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,12 @@ 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; +<<<<<<< HEAD import org.apache.gravitino.exceptions.GravitinoRuntimeException; +======= +import org.apache.gravitino.catalog.mysql.converter.MysqlColumnDefaultValueConverter; +import org.apache.gravitino.catalog.mysql.converter.MysqlTypeConverter; +>>>>>>> 3ce91977a ([#13945] fix(jdbc): Honor pool wait settings and validate MySQL table properties (#13946)) import org.apache.gravitino.rel.Column; import org.apache.gravitino.rel.TableChange; import org.apache.gravitino.rel.expressions.distributions.Distributions; @@ -155,15 +160,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..7cd914c449 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,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(); @@ -147,4 +225,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..ba27f6c56b 100644 --- a/docs/jdbc-clickhouse-catalog.md +++ b/docs/jdbc-clickhouse-catalog.md @@ -48,6 +48,7 @@ Pass any JDBC pool property that Gravitino does not define by adding the `gravit When using the JDBC catalog you must provide `jdbc-url`, `jdbc-driver`, `jdbc-user`, and `jdbc-password`. Common catalog properties are listed [here](./gravitino-server-config.md#catalog-properties-configuration); ClickHouse adds no extra catalog-scoped keys. +<<<<<<< HEAD | Configuration item | Description | Default value | Required | |-------------------------|-----------------------------------------------------------------------|---------------|----------| | `jdbc-url` | JDBC URL, for example `jdbc:clickhouse://localhost:8123` | (none) | Yes | @@ -57,6 +58,18 @@ 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-wait-ms` | Max wait time for a connection | `30000` | No | +======= +| Configuration item | Description | Default value | Required | +|-------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|---------------|----------| +| `jdbc-url` | JDBC URL, for example `jdbc:clickhouse://localhost:8123` | (none) | Yes | +| `jdbc-driver` | JDBC driver class, for example `com.clickhouse.jdbc.ClickHouseDriver` | (none) | Yes | +| `jdbc-user` | JDBC user name | (none) | Yes | +| `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-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 | +>>>>>>> 3ce91977a ([#13945] fix(jdbc): Honor pool wait settings and validate MySQL table properties (#13946)) ### Create a ClickHouse Catalog diff --git a/docs/jdbc-doris-catalog.md b/docs/jdbc-doris-catalog.md index 47bf6e02aa..2ebcc5fa53 100644 --- a/docs/jdbc-doris-catalog.md +++ b/docs/jdbc-doris-catalog.md @@ -41,6 +41,7 @@ more details. Besides the [common catalog properties](./gravitino-server-config.md#catalog-properties-configuration), the Doris catalog has the following properties: +<<<<<<< HEAD | Configuration item | Description | Default value | Required | |-------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|---------------|----------| | `jdbc-url` | JDBC URL for connecting to the database. For example, `jdbc:mysql://localhost:9030` | (none) | Yes | @@ -51,6 +52,18 @@ Besides the [common catalog properties](./gravitino-server-config.md#catalog-pro | `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 | +======= +| Configuration item | Description | Default value | Required | +|-------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|---------------|----------| +| `jdbc-url` | JDBC URL for connecting to the database. For example, `jdbc:mysql://localhost:9030` | (none) | Yes | +| `jdbc-driver` | The driver of the JDBC connection. For example, `com.mysql.jdbc.Driver`. | (none) | Yes | +| `jdbc-user` | The JDBC user name. | (none) | Yes | +| `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-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 | +>>>>>>> 3ce91977a ([#13945] fix(jdbc): Honor pool wait settings and validate MySQL table properties (#13946)) 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 a4412dbd47..60223cacf5 100644 --- a/docs/jdbc-hologres-catalog.md +++ b/docs/jdbc-hologres-catalog.md @@ -43,6 +43,7 @@ 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: +<<<<<<< HEAD | 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 | @@ -52,6 +53,19 @@ Besides the [common catalog properties](./gravitino-server-config.md#catalog-pro | `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 | +======= +| 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 | +>>>>>>> 3ce91977a ([#13945] fix(jdbc): Honor pool wait settings and validate MySQL table properties (#13946)) :::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 bd2964861b..0c8846e8b2 100644 --- a/docs/jdbc-mysql-catalog.md +++ b/docs/jdbc-mysql-catalog.md @@ -41,6 +41,7 @@ When using Gravitino with Trino, pass the Trino MySQL connector configuration us If you use a JDBC catalog, you must provide `jdbc-url`, `jdbc-driver`, `jdbc-user` and `jdbc-password` to catalog properties. Besides the [common catalog properties](./gravitino-server-config.md#catalog-properties-configuration), the MySQL catalog has the following properties: +<<<<<<< HEAD | Configuration item | Description | Default value | Required | |-------------------------|--------------------------------------------------------------------------------------------------------|---------------|----------| | `jdbc-url` | JDBC URL for connecting to the database. For example, `jdbc:mysql://localhost:3306` | (none) | Yes | @@ -50,6 +51,18 @@ 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-wait-ms` | The maximum Duration that the pool will wait for a connection to be returned. `30000` by default. | `30000` | No | +======= +| Configuration item | Description | Default value | Required | +|-------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|---------------|----------| +| `jdbc-url` | JDBC URL for connecting to the database. For example, `jdbc:mysql://localhost:3306` | (none) | Yes | +| `jdbc-driver` | The driver of the JDBC connection. For example, `com.mysql.jdbc.Driver` or `com.mysql.cj.jdbc.Driver`. | (none) | Yes | +| `jdbc-user` | The JDBC user name. | (none) | Yes | +| `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-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 | +>>>>>>> 3ce91977a ([#13945] fix(jdbc): Honor pool wait settings and validate MySQL table properties (#13946)) :::caution Gravitino does not package the MySQL JDBC driver, because MySQL Connector/J is @@ -222,7 +235,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 9e0b99e7a5..0b502dcdcf 100644 --- a/docs/jdbc-oceanbase-catalog.md +++ b/docs/jdbc-oceanbase-catalog.md @@ -39,6 +39,7 @@ 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-user` and `jdbc-password` to catalog properties. Besides the [common catalog properties](./gravitino-server-config.md#catalog-properties-configuration), the OceanBase catalog has the following properties: +<<<<<<< HEAD | Configuration item | Description | Default value | Required | |-------------------------|---------------------------------------------------------------------------------------------------------------------------------------|---------------|----------| | `jdbc-url` | JDBC URL for connecting to the database. For example, `jdbc:mysql://localhost:2881` or `jdbc:oceanbase://localhost:2881` | (none) | Yes | @@ -48,6 +49,18 @@ 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-wait-ms` | The maximum Duration that the pool will wait for a connection to be returned. `30000` by default. | `30000` | No | +======= +| Configuration item | Description | Default value | Required | +|-------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|---------------|----------| +| `jdbc-url` | JDBC URL for connecting to the database. For example, `jdbc:mysql://localhost:2881` or `jdbc:oceanbase://localhost:2881` | (none) | Yes | +| `jdbc-driver` | The driver of the JDBC connection. For example, `com.mysql.jdbc.Driver` or `com.mysql.cj.jdbc.Driver` or `com.oceanbase.jdbc.Driver`. | (none) | Yes | +| `jdbc-user` | The JDBC user name. | (none) | Yes | +| `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-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 | +>>>>>>> 3ce91977a ([#13945] fix(jdbc): Honor pool wait settings and validate MySQL table properties (#13946)) :::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..c94002d1eb 100644 --- a/docs/jdbc-postgresql-catalog.md +++ b/docs/jdbc-postgresql-catalog.md @@ -39,6 +39,7 @@ When using Gravitino with Trino, pass the Trino PostgreSQL connector configurati If you use JDBC catalog, you must provide `jdbc-url`, `jdbc-driver`, `jdbc-user` and `jdbc-password` to catalog properties. Besides the [common catalog properties](./gravitino-server-config.md#catalog-properties-configuration), the PostgreSQL catalog has the following properties: +<<<<<<< HEAD | Configuration item | Description | Default value | Required | |-------------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------|---------------|----------| | `jdbc-url` | A valid PostgreSQL JDBC URL, for example `jdbc:postgresql://localhost:5432/pg_database?sslmode=require`. If the URL omits the database, set `jdbc-database`. | (none) | Yes | @@ -49,6 +50,19 @@ 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-wait-ms` | The maximum Duration that the pool will wait for a connection to be returned. `30000` by default. | `30000` | No | +======= +| Configuration item | Description | Default value | Required | +|-------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|---------------|--------------------------------| +| `jdbc-url` | A valid PostgreSQL JDBC URL, for example `jdbc:postgresql://localhost:5432/pg_database?sslmode=require`. If the URL omits the database, set `jdbc-database`. | (none) | Yes | +| `jdbc-driver` | The driver of the JDBC connection. For example `org.postgresql.Driver`. | (none) | Yes | +| `jdbc-database` | The database of the JDBC connection. Derived from `jdbc-url` when omitted. An explicit value must be nonblank and match the URL database when both are provided. | (none) | Only if absent from `jdbc-url` | +| `jdbc-user` | The JDBC user name. | (none) | Yes | +| `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-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 | +>>>>>>> 3ce91977a ([#13945] fix(jdbc): Honor pool wait settings and validate MySQL table properties (#13946)) :::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..ac0ffa3b4d 100644 --- a/docs/jdbc-starrocks-catalog.md +++ b/docs/jdbc-starrocks-catalog.md @@ -40,6 +40,7 @@ more details. Besides the [common catalog properties](./gravitino-server-config.md#catalog-properties-configuration), the StarRocks catalog has the following properties: +<<<<<<< HEAD | Configuration item | Description | Default value | Required | |-------------------------|---------------------------------------------------------------------------------------------------|---------------|----------| | `jdbc-url` | JDBC URL for connecting to the database. For example, `jdbc:mysql://localhost:9030` | (none) | Yes | @@ -49,6 +50,18 @@ 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-wait-ms` | The maximum Duration that the pool will wait for a connection to be returned. `30000` by default. | `30000` | No | +======= +| Configuration item | Description | Default value | Required | +|-------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|---------------|----------| +| `jdbc-url` | JDBC URL for connecting to the database. For example, `jdbc:mysql://localhost:9030` | (none) | Yes | +| `jdbc-driver` | The driver of the JDBC connection. For example, `com.mysql.jdbc.Driver`. | (none) | Yes | +| `jdbc-user` | The JDBC user name. | (none) | Yes | +| `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-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 | +>>>>>>> 3ce91977a ([#13945] fix(jdbc): Honor pool wait settings and validate MySQL table properties (#13946)) 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..a0b91b456c 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,12 +23,57 @@ 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 +<<<<<<< HEAD +======= + void testGetNoTablePropertiesInstance() { + JdbcPropertiesConverter instance = JdbcPropertiesConverter.getNoTablePropertiesInstance(); + Assertions.assertNotNull(instance); + Assertions.assertSame(instance, JdbcPropertiesConverter.getNoTablePropertiesInstance()); + 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 +>>>>>>> 3ce91977a ([#13945] fix(jdbc): Honor pool wait settings and validate MySQL table properties (#13946)) void testCatalogProperties() { String url = "jdbc-url"; String user = "user1";
