Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions docs/generated/spark_catalog_configuration.html
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,12 @@
<td>String</td>
<td>The default database name.</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 and the default database will not be automatically created during catalog initialization.</td>
</tr>
<tr>
<td><h5>v1Function.enabled</h5></td>
<td style="word-wrap: break-word;">true</td>
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -99,6 +99,7 @@
import static org.apache.paimon.CoreOptions.TYPE;
import static org.apache.paimon.TableType.FORMAT_TABLE;
import static org.apache.paimon.spark.SparkCatalogOptions.DEFAULT_DATABASE;
import static org.apache.paimon.spark.SparkCatalogOptions.DISABLE_CREATE_TABLE_IN_DEFAULT_DB;
import static org.apache.paimon.spark.SparkCatalogOptions.V1FUNCTION_ENABLED;
import static org.apache.paimon.spark.SparkTypeUtils.CURRENT_DEFAULT_COLUMN_METADATA_KEY;
import static org.apache.paimon.spark.SparkTypeUtils.toPaimonType;
Expand Down Expand Up @@ -129,6 +130,7 @@ public class SparkCatalog extends SparkBaseCatalog

private Catalog catalog;
private String defaultDatabase;
private boolean disableCreateTableInDefaultDatabase;
private boolean v1FunctionEnabled;
@Nullable private PaimonV1FunctionRegistry v1FunctionRegistry;

Expand All @@ -144,22 +146,28 @@ public void initialize(String name, CaseInsensitiveStringMap options) {
this.catalog = CatalogFactory.createCatalog(catalogContext);
this.defaultDatabase =
options.getOrDefault(DEFAULT_DATABASE.key(), DEFAULT_DATABASE.defaultValue());
this.disableCreateTableInDefaultDatabase =
options.getBoolean(
DISABLE_CREATE_TABLE_IN_DEFAULT_DB.key(),
DISABLE_CREATE_TABLE_IN_DEFAULT_DB.defaultValue());
this.v1FunctionEnabled =
options.getBoolean(V1FUNCTION_ENABLED.key(), V1FUNCTION_ENABLED.defaultValue())
&& DelegateCatalog.rootCatalog(catalog) instanceof RESTCatalog;
if (v1FunctionEnabled) {
this.v1FunctionRegistry = new PaimonV1FunctionRegistry(sparkSession);
}
try {
catalog.getDatabase(defaultDatabase);
} catch (Catalog.DatabaseNotExistException e) {
LOG.info(
"Default database '{}' does not exist, caused by: {}, start to create it",
defaultDatabase,
ExceptionUtils.stringifyException(e));
if (!disableCreateTableInDefaultDatabase) {
try {
createNamespace(defaultNamespace(), new HashMap<>());
} catch (NamespaceAlreadyExistsException ignored) {
catalog.getDatabase(defaultDatabase);
} catch (Catalog.DatabaseNotExistException e) {
LOG.info(
"Default database '{}' does not exist, caused by: {}, start to create it",
defaultDatabase,
ExceptionUtils.stringifyException(e));
try {
createNamespace(defaultNamespace(), new HashMap<>());
} catch (NamespaceAlreadyExistsException ignored) {
}
}
}
}
Expand Down Expand Up @@ -374,6 +382,12 @@ public org.apache.spark.sql.connector.catalog.Table createTable(
Transform[] partitions,
Map<String, String> properties)
throws TableAlreadyExistsException, NoSuchNamespaceException {
if (disableCreateTableInDefaultDatabase
&& ident.namespace().length == 1
&& ident.namespace()[0].equals(defaultDatabase)) {
throw new UnsupportedOperationException(
"Creating table in default database is disabled, please specify a database name.");
}
try {
catalog.createTable(
toIdentifier(ident, catalogName),
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,14 @@ public class SparkCatalogOptions {
.defaultValue(Catalog.DEFAULT_DATABASE)
.withDescription("The default database name.");

public static final ConfigOption<Boolean> DISABLE_CREATE_TABLE_IN_DEFAULT_DB =
key("disable-create-table-in-default-db")
.booleanType()
.defaultValue(false)
.withDescription(
"If true, creating table in default database is not allowed "
+ "and the default database will not be automatically created during catalog initialization.");

public static final ConfigOption<Boolean> V1FUNCTION_ENABLED =
key("v1Function.enabled")
.booleanType()
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,145 @@
/*
* 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.paimon.spark;

import org.apache.paimon.fs.Path;
import org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions;

import org.apache.spark.sql.SparkSession;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatCode;
import static org.assertj.core.api.Assertions.assertThatThrownBy;

/** Tests for {@code disable-create-table-in-default-db} option in Spark. */
public class SparkCatalogDisableDefaultDbTest {

private SparkSession spark;

@AfterEach
public void stopSpark() {
if (spark != null) {
spark.stop();
spark = null;
}
}

@Test
public void testDisableCreateTableInDefaultDb(@TempDir java.nio.file.Path tempDir) {
Path warehousePath = new Path("file:" + tempDir.toString());
spark =
SparkSession.builder()
.master("local[2]")
.config("spark.sql.catalog.paimon", SparkCatalog.class.getName())
.config("spark.sql.catalog.paimon.warehouse", warehousePath.toString())
.config(
"spark.sql.catalog.paimon.disable-create-table-in-default-db",
"true")
.config(
"spark.sql.extensions",
PaimonSparkSessionExtensions.class.getName())
.getOrCreate();

// Creating table in default database should fail
assertThatThrownBy(
() ->
spark.sql(
"CREATE TABLE paimon.default.t1 (a INT, b STRING) USING paimon"))
.hasMessageContaining(
"Creating table in default database is disabled, please specify a database name.");

// Creating a non-default database should succeed
assertThatCode(() -> spark.sql("CREATE DATABASE paimon.my_db")).doesNotThrowAnyException();

// Creating table in a non-default database should succeed
assertThatCode(
() ->
spark.sql(
"CREATE TABLE paimon.my_db.t1 (a INT, b STRING) USING paimon"))
.doesNotThrowAnyException();

// Verify the table is accessible
spark.sql("INSERT INTO paimon.my_db.t1 VALUES (1, 'hello')").collect();
assertThat(
spark.sql("SELECT * FROM paimon.my_db.t1").collectAsList().stream()
.map(Object::toString))
.containsExactly("[1,hello]");
}

@Test
public void testDisableCreateTableWithCustomDefaultDb(@TempDir java.nio.file.Path tempDir) {
Path warehousePath = new Path("file:" + tempDir.toString());
spark =
SparkSession.builder()
.master("local[2]")
.config("spark.sql.catalog.paimon", SparkCatalog.class.getName())
.config("spark.sql.catalog.paimon.warehouse", warehousePath.toString())
.config(
"spark.sql.catalog.paimon.disable-create-table-in-default-db",
"true")
.config("spark.sql.catalog.paimon.defaultDatabase", "custom_default")
.config(
"spark.sql.extensions",
PaimonSparkSessionExtensions.class.getName())
.getOrCreate();

// Creating table in custom default database should fail
assertThatThrownBy(
() ->
spark.sql(
"CREATE TABLE paimon.custom_default.t1 (a INT, b STRING) USING paimon"))
.hasMessageContaining(
"Creating table in default database is disabled, please specify a database name.");

// Creating a different database and table should succeed
assertThatCode(() -> spark.sql("CREATE DATABASE paimon.other_db"))
.doesNotThrowAnyException();
assertThatCode(
() ->
spark.sql(
"CREATE TABLE paimon.other_db.t1 (a INT, b STRING) USING paimon"))
.doesNotThrowAnyException();
}

@Test
public void testDefaultDatabaseNotCreatedWhenDisabled(@TempDir java.nio.file.Path tempDir) {
Path warehousePath = new Path("file:" + tempDir.toString());
spark =
SparkSession.builder()
.master("local[2]")
.config("spark.sql.catalog.paimon", SparkCatalog.class.getName())
.config("spark.sql.catalog.paimon.warehouse", warehousePath.toString())
.config(
"spark.sql.catalog.paimon.disable-create-table-in-default-db",
"true")
.config(
"spark.sql.extensions",
PaimonSparkSessionExtensions.class.getName())
.getOrCreate();

// Default database should not have been auto-created
assertThat(
spark.sql("SHOW DATABASES IN paimon").collectAsList().stream()
.map(r -> r.getString(0)))
.doesNotContain("default");
}
}
Loading