This is an automated email from the ASF dual-hosted git repository.
zjureel pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 2c87df057 [flink] add option to support disable creating table in
default database (#1734)
2c87df057 is described below
commit 2c87df057bee2106855c96c1f0eee73cc604481a
Author: legendtkl <[email protected]>
AuthorDate: Tue Sep 12 14:55:04 2023 +0800
[flink] add option to support disable creating table in default database
(#1734)
Co-authored-by: Shammon FY <[email protected]>
---
.../generated/catalog_configuration.html | 6 +++
.../generated/flink_catalog_configuration.html | 6 +++
.../java/org/apache/paimon/flink/FlinkCatalog.java | 11 +++++
.../apache/paimon/flink/FlinkCatalogOptions.java | 6 +++
.../org/apache/paimon/flink/FlinkCatalogTest.java | 53 ++++++++++++++++++++++
5 files changed, 82 insertions(+)
diff --git a/docs/layouts/shortcodes/generated/catalog_configuration.html
b/docs/layouts/shortcodes/generated/catalog_configuration.html
index d243737a8..5b84e4854 100644
--- a/docs/layouts/shortcodes/generated/catalog_configuration.html
+++ b/docs/layouts/shortcodes/generated/catalog_configuration.html
@@ -26,6 +26,12 @@ under the License.
</tr>
</thead>
<tbody>
+ <tr>
+ <td><h5>disable-create-table-in-default-db</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>If true, creating table in default database is not allowed.
Default is false.</td>
+ </tr>
<tr>
<td><h5>fs.allow-hadoop-fallback</h5></td>
<td style="word-wrap: break-word;">true</td>
diff --git a/docs/layouts/shortcodes/generated/flink_catalog_configuration.html
b/docs/layouts/shortcodes/generated/flink_catalog_configuration.html
index 15d80eac7..7ff460e84 100644
--- a/docs/layouts/shortcodes/generated/flink_catalog_configuration.html
+++ b/docs/layouts/shortcodes/generated/flink_catalog_configuration.html
@@ -32,6 +32,12 @@ under the License.
<td>String</td>
<td></td>
</tr>
+ <tr>
+ <td><h5>disable-create-table-in-default-db</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>If true, creating table in default database is not allowed.
Default is false.</td>
+ </tr>
<tr>
<td><h5>log.system.auto-register</h5></td>
<td style="word-wrap: break-word;">false</td>
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java
index 23cf7a9ba..c52b75984 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java
@@ -91,6 +91,7 @@ import static
org.apache.flink.table.descriptors.Schema.SCHEMA;
import static org.apache.flink.table.factories.FactoryUtil.CONNECTOR;
import static
org.apache.flink.table.types.utils.TypeConversions.fromLogicalToDataType;
import static org.apache.paimon.CoreOptions.PATH;
+import static
org.apache.paimon.flink.FlinkCatalogOptions.DISABLE_CREATE_TABLE_IN_DEFAULT_DB;
import static
org.apache.paimon.flink.FlinkCatalogOptions.LOG_SYSTEM_AUTO_REGISTER;
import static org.apache.paimon.flink.FlinkCatalogOptions.REGISTER_TIMEOUT;
import static org.apache.paimon.flink.LogicalTypeConversion.toDataType;
@@ -114,6 +115,8 @@ public class FlinkCatalog extends AbstractCatalog {
private final Duration logStoreAutoRegisterTimeout;
+ private final boolean disableCreateTableInDefaultDatabase;
+
public FlinkCatalog(
Catalog catalog,
String name,
@@ -125,6 +128,7 @@ public class FlinkCatalog extends AbstractCatalog {
this.classLoader = classLoader;
this.logStoreAutoRegister = options.get(LOG_SYSTEM_AUTO_REGISTER);
this.logStoreAutoRegisterTimeout = options.get(REGISTER_TIMEOUT);
+ this.disableCreateTableInDefaultDatabase =
options.get(DISABLE_CREATE_TABLE_IN_DEFAULT_DB);
try {
this.catalog.createDatabase(defaultDatabase, true);
} catch (Catalog.DatabaseAlreadyExistException ignore) {
@@ -251,6 +255,13 @@ public class FlinkCatalog extends AbstractCatalog {
throw new UnsupportedOperationException(
"Only support CatalogTable, but is: " + table.getClass());
}
+
+ if (getDefaultDatabase().equals(tablePath.getDatabaseName())
+ && disableCreateTableInDefaultDatabase) {
+ throw new UnsupportedOperationException(
+ "Creating table in default database is disabled, please
specify a database name.");
+ }
+
CatalogTable catalogTable = (CatalogTable) table;
Map<String, String> options = table.getOptions();
String connector = options.get(CONNECTOR.key());
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalogOptions.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalogOptions.java
index 97c1d5424..022495385 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalogOptions.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalogOptions.java
@@ -47,4 +47,10 @@ public class FlinkCatalogOptions {
.defaultValue(Duration.ofMinutes(1))
.withDescription(
"The timeout for register to create or delete
topic in log system.");
+ public static final ConfigOption<Boolean>
DISABLE_CREATE_TABLE_IN_DEFAULT_DB =
+ ConfigOptions.key("disable-create-table-in-default-db")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "If true, creating table in default database is
not allowed. Default is false.");
}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkCatalogTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkCatalogTest.java
index 2bd0efa21..0e69c2672 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkCatalogTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkCatalogTest.java
@@ -70,9 +70,11 @@ import java.util.Map;
import java.util.UUID;
import java.util.stream.Stream;
+import static
org.apache.paimon.flink.FlinkCatalogOptions.DISABLE_CREATE_TABLE_IN_DEFAULT_DB;
import static
org.apache.paimon.flink.FlinkCatalogOptions.LOG_SYSTEM_AUTO_REGISTER;
import static org.apache.paimon.flink.FlinkConnectorOptions.LOG_SYSTEM;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Test for {@link FlinkCatalog}. */
@@ -81,6 +83,10 @@ public class FlinkCatalogTest {
private final ObjectPath path1 = new ObjectPath("db1", "t1");
private final ObjectPath path3 = new ObjectPath("db1", "t2");
+
+ private final ObjectPath tableInDefaultDb = new ObjectPath("default",
"t1");
+
+ private final ObjectPath tableInDefaultDb1 = new ObjectPath("default-db",
"t1");
private final ObjectPath nonExistDbPath =
ObjectPath.fromString("non.exist");
private final ObjectPath nonExistObjectPath =
ObjectPath.fromString("db1.nonexist");
private Catalog catalog;
@@ -504,6 +510,53 @@ public class FlinkCatalogTest {
.hasMessage("Check unregister log store topic here.");
}
+ @Test
+ public void testDisableCreateTableInDefaultDB()
+ throws TableAlreadyExistException, DatabaseNotExistException,
+ DatabaseAlreadyExistException {
+ String path = new File(temporaryFolder.toFile(),
UUID.randomUUID().toString()).toString();
+ Options conf = new Options();
+ conf.setString("warehouse", path);
+ conf.set(LOG_SYSTEM_AUTO_REGISTER, true);
+ conf.set(DISABLE_CREATE_TABLE_IN_DEFAULT_DB, true);
+ Catalog catalog =
+ FlinkCatalogFactory.createCatalog(
+ "test-ddl-catalog",
+ CatalogContext.create(conf),
+ FlinkCatalogTest.class.getClassLoader());
+
+ assertThatThrownBy(
+ () ->
+ catalog.createTable(
+ tableInDefaultDb,
+ this.createTable(new HashMap<>(0)),
+ false))
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessage(
+ "Creating table in default database is disabled,
please specify a database name.");
+
+ catalog.createDatabase("db1", null, false);
+ assertThatCode(() -> catalog.createTable(path1, this.createTable(new
HashMap<>(0)), false))
+ .doesNotThrowAnyException();
+
+ conf.set(FlinkCatalogOptions.DEFAULT_DATABASE, "default-db");
+ Catalog catalog1 =
+ FlinkCatalogFactory.createCatalog(
+ "test-ddl-catalog1",
+ CatalogContext.create(conf),
+ FlinkCatalogTest.class.getClassLoader());
+
+ assertThatThrownBy(
+ () ->
+ catalog1.createTable(
+ tableInDefaultDb1,
+ this.createTable(new HashMap<>(0)),
+ false))
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessage(
+ "Creating table in default database is disabled,
please specify a database name.");
+ }
+
private void checkEquals(ObjectPath path, CatalogTable t1, CatalogTable
t2) {
Path tablePath =
((AbstractCatalog) ((FlinkCatalog) catalog).catalog())
