This is an automated email from the ASF dual-hosted git repository.
diqiu50 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 0e04956518 [#13798] feat(flink): Support JDBC catalogs without a
default database (#13876)
0e04956518 is described below
commit 0e049565183f9f9da6492d894ceebeb28744ac1b
Author: Shabbir Hussain <[email protected]>
AuthorDate: Sun Oct 25 19:41:58 2026 -0700
[#13798] feat(flink): Support JDBC catalogs without a default database
(#13876)
### What changes were proposed in this pull request?
Allow the Flink JDBC catalog to initialize when
`flink.bypass.default-database` is not configured.
Fully qualified table references continue to resolve through Gravitino,
while operations requiring a native default-database-backed JDBC catalog
return a clear error. Statistics requests return `UNKNOWN` when no
default database is configured.
Add unit coverage for the no-default-database behavior and a
Docker-backed MySQL integration test for a fully qualified query.
### Why are the changes needed?
Flink previously rejected JDBC catalogs without
`flink.bypass.default-database`, even when queries specified the
complete `catalog.schema.table` path.
This change supports MySQL catalogs without a default database as
requested.
Fix: #13798
### Does this PR introduce _any_ user-facing change?
Yes. Flink users can now use Gravitino JDBC/MySQL catalogs without
configuring `flink.bypass.default-database` when they use fully
qualified table names.
Operations that require a default database now fail with a clear
message.
### How was this patch tested?
- Added common unit coverage for opening without a default database,
`UNKNOWN` statistics, and the clear default-dependent error path.
- Ran the focused JDBC catalog tests across Flink 1.19, 1.20, 2.1, and
2.2 (five tests passed on each).
- Ran the Docker-backed Flink 1.20 MySQL regression test
`FlinkJdbcMysqlCatalogIT120.testSelectFullyQualifiedTableWithoutDefaultDatabase`.
---
.../flink/connector/jdbc/GravitinoJdbcCatalog.java | 58 ++++++++++++++++++++--
.../jdbc/GravitinoJdbcCatalogFactory.java | 7 +--
.../test/jdbc/FlinkJdbcMysqlCatalogIT.java | 44 ++++++++++++++++
.../connector/jdbc/TestGravitinoJdbcCatalog.java | 47 ++++++++++++++++--
4 files changed, 143 insertions(+), 13 deletions(-)
diff --git
a/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/jdbc/GravitinoJdbcCatalog.java
b/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/jdbc/GravitinoJdbcCatalog.java
index 41242d7447..0e14335a9a 100644
---
a/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/jdbc/GravitinoJdbcCatalog.java
+++
b/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/jdbc/GravitinoJdbcCatalog.java
@@ -23,9 +23,14 @@ import com.google.common.annotations.VisibleForTesting;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
+import javax.annotation.Nullable;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.table.catalog.AbstractCatalog;
+import org.apache.flink.table.catalog.ObjectPath;
import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.TableNotExistException;
+import org.apache.flink.table.catalog.stats.CatalogColumnStatistics;
+import org.apache.flink.table.catalog.stats.CatalogTableStatistics;
import org.apache.flink.table.factories.CatalogFactory;
import org.apache.flink.table.factories.Factory;
import org.apache.gravitino.Catalog;
@@ -42,7 +47,10 @@ import
org.apache.gravitino.flink.connector.catalog.BaseCatalog;
*/
public abstract class GravitinoJdbcCatalog extends BaseCatalog {
+ private static final String NO_DEFAULT_DATABASE =
"__gravitino_no_default_database__";
+
private final CatalogFactory.Context context;
+ @Nullable private final String configuredDefaultDatabase;
private AbstractCatalog jdbcCatalog;
// Mutable copy shared with BaseCatalog.catalogOptions so credential
injection in open() is
// visible to toFlinkTable() and to the inner JdbcCatalogFactory context.
@@ -50,7 +58,7 @@ public abstract class GravitinoJdbcCatalog extends
BaseCatalog {
protected GravitinoJdbcCatalog(
CatalogFactory.Context context,
- String defaultDatabase,
+ @Nullable String defaultDatabase,
SchemaAndTablePropertiesConverter schemaAndTablePropertiesConverter,
PartitionConverter partitionConverter) {
this(context, defaultDatabase, schemaAndTablePropertiesConverter,
partitionConverter, null);
@@ -58,7 +66,7 @@ public abstract class GravitinoJdbcCatalog extends
BaseCatalog {
protected GravitinoJdbcCatalog(
CatalogFactory.Context context,
- String defaultDatabase,
+ @Nullable String defaultDatabase,
SchemaAndTablePropertiesConverter schemaAndTablePropertiesConverter,
PartitionConverter partitionConverter,
AbstractCatalog jdbcCatalog) {
@@ -73,7 +81,7 @@ public abstract class GravitinoJdbcCatalog extends
BaseCatalog {
private GravitinoJdbcCatalog(
CatalogFactory.Context context,
- String defaultDatabase,
+ @Nullable String defaultDatabase,
SchemaAndTablePropertiesConverter schemaAndTablePropertiesConverter,
PartitionConverter partitionConverter,
AbstractCatalog jdbcCatalog,
@@ -81,10 +89,11 @@ public abstract class GravitinoJdbcCatalog extends
BaseCatalog {
super(
context.getName(),
mutableOptions,
- defaultDatabase,
+ defaultDatabase == null ? NO_DEFAULT_DATABASE : defaultDatabase,
schemaAndTablePropertiesConverter,
partitionConverter);
this.context = context;
+ this.configuredDefaultDatabase = defaultDatabase;
this.jdbcCatalog = jdbcCatalog;
this.mutableOptions = mutableOptions;
}
@@ -101,6 +110,9 @@ public abstract class GravitinoJdbcCatalog extends
BaseCatalog {
// During CREATE CATALOG, open() is called before the catalog is stored
in Gravitino.
// In this case credentials are already present in the user-provided
options.
}
+ if (configuredDefaultDatabase == null) {
+ return;
+ }
CatalogFactory.Context contextWithCredentials =
new CatalogFactory.Context() {
@Override
@@ -127,6 +139,20 @@ public abstract class GravitinoJdbcCatalog extends
BaseCatalog {
super.open();
}
+ /** Closes the native JDBC catalog when one was created for a configured
default database. */
+ @Override
+ public void close() throws CatalogException {
+ if (jdbcCatalog != null) {
+ super.close();
+ }
+ }
+
+ @Override
+ @Nullable
+ public String getDefaultDatabase() {
+ return configuredDefaultDatabase;
+ }
+
/**
* Creates the inner Flink JDBC catalog from the given context. Subclasses
for different Flink
* versions implement this because {@code JdbcCatalogFactory}'s package
moved (and, starting with
@@ -139,9 +165,33 @@ public abstract class GravitinoJdbcCatalog extends
BaseCatalog {
@Override
protected AbstractCatalog realCatalog() {
+ if (jdbcCatalog == null) {
+ throw new CatalogException(
+ "No default database is configured for catalog '"
+ + getName()
+ + "'. Select a database before using operations that require a
default database.");
+ }
return jdbcCatalog;
}
+ @Override
+ public CatalogTableStatistics getTableStatistics(ObjectPath tablePath)
+ throws TableNotExistException, CatalogException {
+ if (configuredDefaultDatabase == null) {
+ return CatalogTableStatistics.UNKNOWN;
+ }
+ return super.getTableStatistics(tablePath);
+ }
+
+ @Override
+ public CatalogColumnStatistics getTableColumnStatistics(ObjectPath tablePath)
+ throws TableNotExistException, CatalogException {
+ if (configuredDefaultDatabase == null) {
+ return CatalogColumnStatistics.UNKNOWN;
+ }
+ return super.getTableColumnStatistics(tablePath);
+ }
+
@Override
public Optional<Factory> getFactory() {
return Optional.of(catalogCompat().jdbcDynamicTableFactory());
diff --git
a/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/jdbc/GravitinoJdbcCatalogFactory.java
b/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/jdbc/GravitinoJdbcCatalogFactory.java
index f6a1dec89f..632326408a 100644
---
a/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/jdbc/GravitinoJdbcCatalogFactory.java
+++
b/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/jdbc/GravitinoJdbcCatalogFactory.java
@@ -21,10 +21,10 @@ package org.apache.gravitino.flink.connector.jdbc;
import java.util.Collections;
import java.util.Set;
+import javax.annotation.Nullable;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.table.catalog.Catalog;
import org.apache.flink.table.factories.FactoryUtil;
-import org.apache.flink.util.Preconditions;
import org.apache.gravitino.flink.connector.PartitionConverter;
import org.apache.gravitino.flink.connector.SchemaAndTablePropertiesConverter;
import org.apache.gravitino.flink.connector.UnsupportPartitionConverter;
@@ -48,9 +48,6 @@ public abstract class GravitinoJdbcCatalogFactory implements
BaseCatalogFactory
FactoryUtils.createCatalogFactoryHelper(this, context);
String defaultDatabase =
helper.getOptions().get(GravitinoJdbcCatalogFactoryOptions.DEFAULT_DATABASE);
- Preconditions.checkArgument(
- defaultDatabase != null,
- GravitinoJdbcCatalogFactoryOptions.DEFAULT_DATABASE.key() + " should
not be null.");
return newCatalog(
context, defaultDatabase, schemaAndTablePropertiesConverter(),
partitionConverter());
}
@@ -59,7 +56,7 @@ public abstract class GravitinoJdbcCatalogFactory implements
BaseCatalogFactory
// every concrete, version-specific factory overrides this hook.
protected abstract Catalog newCatalog(
Context context,
- String defaultDatabase,
+ @Nullable String defaultDatabase,
SchemaAndTablePropertiesConverter schemaAndTablePropertiesConverter,
PartitionConverter partitionConverter);
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 77f0a4ea4b..d265bffe23 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
@@ -25,6 +25,7 @@ import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableMap;
import java.util.Map;
import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.types.Row;
import org.apache.gravitino.Catalog;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.credential.CredentialConstants;
@@ -196,6 +197,49 @@ public abstract class FlinkJdbcMysqlCatalogIT extends
FlinkCommonIT {
Assertions.assertEquals(driver,
properties.get(JdbcPropertiesConstants.GRAVITINO_JDBC_DRIVER));
}
+ @Test
+ public void testSelectFullyQualifiedTableWithoutDefaultDatabase() throws
Exception {
+ String catalogName =
"gravitino_mysql_jdbc_catalog_without_default_database";
+ String schemaName = "test_no_default_database";
+ String tableName = "test_fully_qualified_table";
+ Catalog catalogWithoutDefaultDatabase = null;
+ try {
+ catalog.asSchemas().createSchema(schemaName, null, ImmutableMap.of());
+ tableEnv.useCatalog(CATALOG_NAME);
+ tableEnv.useDatabase(schemaName);
+ tableEnv.executeSql("CREATE TABLE " + tableName + " (id INT)");
+ tableEnv.executeSql("INSERT INTO " + tableName + " VALUES (1)");
+
+ catalogWithoutDefaultDatabase =
+ metalake.createCatalog(
+ catalogName,
+ org.apache.gravitino.Catalog.Type.RELATIONAL,
+ getProvider(),
+ null,
+ ImmutableMap.of(
+ JdbcPropertiesConstants.GRAVITINO_JDBC_USER,
+ mysqlUsername,
+ JdbcPropertiesConstants.GRAVITINO_JDBC_PASSWORD,
+ mysqlPassword,
+ JdbcPropertiesConstants.GRAVITINO_JDBC_URL,
+ mysqlUrl,
+ JdbcPropertiesConstants.GRAVITINO_JDBC_DRIVER,
+ mysqlDriver));
+
+ Assertions.assertEquals(
+ Row.of(1),
+ tableEnv
+ .executeSql("SELECT * FROM " + catalogName + "." + schemaName +
"." + tableName)
+ .collect()
+ .next());
+ } finally {
+ if (catalogWithoutDefaultDatabase != null) {
+ metalake.dropCatalog(catalogName, true);
+ }
+ catalog.asSchemas().dropSchema(schemaName, true);
+ }
+ }
+
@Test
public void testCreateGravitinoJdbcCatalogUsingSQLMissingOptions() {
tableEnv.useCatalog(DEFAULT_CATALOG);
diff --git
a/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/jdbc/TestGravitinoJdbcCatalog.java
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/jdbc/TestGravitinoJdbcCatalog.java
index b1a3c86d93..d65d632bc2 100644
---
a/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/jdbc/TestGravitinoJdbcCatalog.java
+++
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/jdbc/TestGravitinoJdbcCatalog.java
@@ -26,6 +26,10 @@ import java.util.Optional;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.table.catalog.AbstractCatalog;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.stats.CatalogColumnStatistics;
+import org.apache.flink.table.catalog.stats.CatalogTableStatistics;
import org.apache.flink.table.factories.CatalogFactory;
import org.apache.flink.table.factories.Factory;
import org.apache.gravitino.Audit;
@@ -66,14 +70,35 @@ public abstract class TestGravitinoJdbcCatalog {
catalogCompat().jdbcDynamicTableFactory().getClass(),
factory.get().getClass());
}
+ @Test
+ void testCatalogWithoutDefaultDatabase() throws Exception {
+ TestableGravitinoJdbcCatalog catalog =
+ new TestableGravitinoJdbcCatalog(catalogCompat(), null,
Collections.emptyMap());
+
+ Assertions.assertNull(catalog.getDefaultDatabase());
+ Assertions.assertDoesNotThrow(catalog::open);
+ Assertions.assertFalse(catalog.isInnerCatalogCreated());
+ Assertions.assertEquals(
+ CatalogTableStatistics.UNKNOWN,
+ catalog.getTableStatistics(ObjectPath.fromString("database.table")));
+ Assertions.assertEquals(
+ CatalogColumnStatistics.UNKNOWN,
+
catalog.getTableColumnStatistics(ObjectPath.fromString("database.table")));
+
+ CatalogException exception =
+ Assertions.assertThrows(CatalogException.class,
catalog::getInnerCatalog);
+ Assertions.assertTrue(exception.getMessage().contains("No default database
is configured"));
+ }
+
private static class TestableGravitinoJdbcCatalog extends
GravitinoJdbcCatalog {
private final CatalogCompat catalogCompat;
+ private boolean innerCatalogCreated;
TestableGravitinoJdbcCatalog(
- CatalogCompat catalogCompat, String catalogName, Map<String, String>
options) {
+ CatalogCompat catalogCompat, String defaultDatabase, Map<String,
String> options) {
super(
- new MockCatalogContext(catalogName, options),
- "default",
+ new MockCatalogContext("test", options),
+ defaultDatabase,
Mockito.mock(SchemaAndTablePropertiesConverter.class),
Mockito.mock(PartitionConverter.class));
this.catalogCompat = catalogCompat;
@@ -86,7 +111,21 @@ public abstract class TestGravitinoJdbcCatalog {
@Override
protected AbstractCatalog createInnerCatalog(CatalogFactory.Context
context) {
- throw new UnsupportedOperationException("not exercised by this test");
+ innerCatalogCreated = true;
+ throw new UnsupportedOperationException("The inner catalog must not be
created");
+ }
+
+ @Override
+ protected Catalog catalog() {
+ return new PlainCatalogStub();
+ }
+
+ boolean isInnerCatalogCreated() {
+ return innerCatalogCreated;
+ }
+
+ AbstractCatalog getInnerCatalog() {
+ return realCatalog();
}
}
