This is an automated email from the ASF dual-hosted git repository.
jerryshao 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 27201d342e [#14524] fix(postgresql): preserve schema identity and
contain schema drops (#14528)
27201d342e is described below
commit 27201d342e6e289db00dda8a639da01c1f49d82b
Author: Qi Yu <[email protected]>
AuthorDate: Tue Nov 3 10:05:29 2026 +0800
[#14524] fix(postgresql): preserve schema identity and contain schema drops
(#14528)
### What changes were proposed in this pull request?
Reject a copied marker owned by another live schema before metadata
takeover; preserve genuine external rename through guarded store update
and use no-overwrite insertion for fresh imports. Use atomic PostgreSQL
DROP SCHEMA RESTRICT and conservatively refuse every non-empty schema
even when cascade=true. Align Java/Python contracts, dispatcher Javadoc
and OpenAPI GET/DELETE 409 responses.
### Why are the changes needed?
Loading a copy must not move the original schema's identity metadata,
and dropping a schema must not destroy external dependencies.
Fixes #14524
Fixes #14492
Consolidates #14518 into this PR.
### Does this PR introduce _any_ user-facing change?
Conflicting copied markers are refused; remove the copied source marker
before independent import. Automatic independent copy import is not
implemented. PostgreSQL non-empty schemas return conflict even with
cascade; empty schemas can be dropped. Cold marker imports scan schemas
in the current namespace. No migration or multi-node/cross-system
atomicity claim.
### How was this patch tested?
Full coreUnitTest: 2,692 cases, 2 skipped; PostgreSQL: 79, Docker
enabled, no failures. H2/PostgreSQL duplicate-ID storage contracts,
prior copied-marker mutation/restored-green evidence, and exact-head
independent cumulative review verified. Required ./gradlew :docs:build
passed all three OpenAPI specifications. Deployed multi-node/REST
end-to-end is not claimed.
---
.../java/org/apache/gravitino/SupportsSchemas.java | 16 +-
.../operation/PostgreSqlSchemaOperations.java | 52 ++++-
.../integration/test/CatalogPostgreSqlIT.java | 8 +-
.../operation/TestPostgreSqlSchemaOperations.java | 251 +++++++++++++++++++++
.../apache/gravitino/client/BaseSchemaCatalog.java | 10 +-
.../gravitino/api/supports_schemas.py | 18 +-
.../gravitino/client/base_schema_catalog.py | 11 +-
.../org/apache/gravitino/StringIdentifier.java | 6 +-
.../catalog/SchemaOperationDispatcher.java | 58 ++++-
.../gravitino/connector/SupportsSchemas.java | 16 +-
.../org/apache/gravitino/TestStringIdentifier.java | 9 +
.../catalog/TestSchemaOperationDispatcher.java | 62 +++++
.../relational/service/TestSchemaMetaService.java | 14 ++
docs/jdbc-postgresql-catalog.md | 2 +-
docs/open-api/schemas.yaml | 16 ++
.../testsets/jdbc-postgresql/00006_datatype.sql | 5 +-
.../testsets/jdbc-postgresql/00006_datatype.txt | 3 +
.../trino/connector/GravitinoConnector.java | 8 +
.../TestGravitinoConnectorTransactions.java | 100 ++++++++
19 files changed, 632 insertions(+), 33 deletions(-)
diff --git a/api/src/main/java/org/apache/gravitino/SupportsSchemas.java
b/api/src/main/java/org/apache/gravitino/SupportsSchemas.java
index cc059d9f09..2da71814ba 100644
--- a/api/src/main/java/org/apache/gravitino/SupportsSchemas.java
+++ b/api/src/main/java/org/apache/gravitino/SupportsSchemas.java
@@ -79,6 +79,8 @@ public interface SupportsSchemas {
*
* @param schemaName The name of the schema.
* @return True if the schema exists, false otherwise.
+ * @throws SchemaAlreadyExistsException If loading an external schema
detects an identifier owned
+ * by another existing schema; the conflict is propagated instead of
returning false.
*/
default boolean schemaExists(String schemaName) {
try {
@@ -145,6 +147,8 @@ public interface SupportsSchemas {
* @param schemaName The name of the schema.
* @return A schema.
* @throws NoSuchSchemaException If the schema does not exist (optional).
+ * @throws SchemaAlreadyExistsException If an external schema carries the
identifier of another
+ * existing schema. Remove the copied source marker before importing an
independent copy.
*/
Schema loadSchema(String schemaName) throws NoSuchSchemaException;
@@ -159,17 +163,21 @@ public interface SupportsSchemas {
Schema alterSchema(String schemaName, SchemaChange... changes) throws
NoSuchSchemaException;
/**
- * Drop a schema from the catalog. If cascade option is true, recursively
drop all objects within
- * the schema.
+ * Drop a schema from the catalog. If cascade is true and supported by the
catalog, recursively
+ * drop all objects within the schema.
+ *
+ * <p>A catalog may refuse a non-empty schema even when cascade is requested
if it cannot safely
+ * restrict deletion to objects within that schema. PostgreSQL currently
refuses such requests.
*
* <p>If the catalog implementation does not support this operation, it may
throw {@link
* UnsupportedOperationException}.
*
* @param schemaName The name of the schema.
- * @param cascade If true, recursively drop all objects within the schema.
+ * @param cascade Whether to request recursive deletion, subject to catalog
support.
* @return True if the schema exists and is dropped successfully, false if
the schema doesn't
* exist.
- * @throws NonEmptySchemaException If the schema is not empty and cascade is
false.
+ * @throws NonEmptySchemaException If the schema is non-empty and cascade is
false, or the catalog
+ * cannot safely cascade within the schema.
*/
boolean dropSchema(String schemaName, boolean cascade) throws
NonEmptySchemaException;
}
diff --git
a/catalogs/catalog-jdbc-postgresql/src/main/java/org/apache/gravitino/catalog/postgresql/operation/PostgreSqlSchemaOperations.java
b/catalogs/catalog-jdbc-postgresql/src/main/java/org/apache/gravitino/catalog/postgresql/operation/PostgreSqlSchemaOperations.java
index 57d62c7457..410ae53155 100644
---
a/catalogs/catalog-jdbc-postgresql/src/main/java/org/apache/gravitino/catalog/postgresql/operation/PostgreSqlSchemaOperations.java
+++
b/catalogs/catalog-jdbc-postgresql/src/main/java/org/apache/gravitino/catalog/postgresql/operation/PostgreSqlSchemaOperations.java
@@ -40,12 +40,16 @@ import org.apache.gravitino.catalog.jdbc.JdbcSchema;
import org.apache.gravitino.catalog.jdbc.config.JdbcConfig;
import org.apache.gravitino.catalog.jdbc.converter.JdbcExceptionConverter;
import org.apache.gravitino.catalog.jdbc.operation.JdbcDatabaseOperations;
+import org.apache.gravitino.catalog.jdbc.utils.JdbcConnectorUtils;
import org.apache.gravitino.exceptions.NoSuchSchemaException;
+import org.apache.gravitino.exceptions.NonEmptySchemaException;
import org.apache.gravitino.meta.AuditInfo;
/** Database operations for PostgreSQL. */
public class PostgreSqlSchemaOperations extends JdbcDatabaseOperations {
+ private static final String DEPENDENT_OBJECTS_STILL_EXIST = "2BP01";
+
private String database;
@Override
@@ -142,14 +146,52 @@ public class PostgreSqlSchemaOperations extends
JdbcDatabaseOperations {
return metaData.getSchemas(connection.getCatalog(), pattern);
}
+ /**
+ * Generates an atomic, restrictive schema drop even when cascade is
requested.
+ *
+ * <p>PostgreSQL cascade can remove objects outside this schema. Until
schema-local cascade is
+ * supported safely, the generated statement refuses every non-empty schema.
+ *
+ * @param schema the schema to drop
+ * @param cascade whether cascade was requested; it does not enable
PostgreSQL native cascade
+ * @return the restrictive drop statement
+ */
@Override
public String generateDropDatabaseSql(String schema, boolean cascade) {
- StringBuilder sqlBuilder =
- new StringBuilder(String.format("DROP SCHEMA %s%s%s", PG_QUOTE,
schema, PG_QUOTE));
- if (cascade) {
- sqlBuilder.append(" CASCADE");
+ // PostgreSQL CASCADE also removes objects outside the schema. A
client-side dependency
+ // preflight cannot close the race with concurrent DDL, so always let
PostgreSQL enforce
+ // RESTRICT atomically. Non-empty cascade requests are refused until a
schema-local cascade
+ // can be guaranteed for every object kind and dependency type.
+ return "DROP SCHEMA " + quoteName(schema) + " RESTRICT";
+ }
+
+ /**
+ * Drops an empty schema without requiring explicit locks on its relations.
+ *
+ * @param schema the schema to drop
+ * @param cascade whether cascade was requested
+ * @throws NonEmptySchemaException if the schema is non-empty, including
when cascade is requested
+ */
+ @Override
+ protected void dropDatabase(String schema, boolean cascade) {
+ try (Connection connection = getConnection()) {
+ JdbcConnectorUtils.executeUpdate(connection,
generateDropDatabaseSql(schema, cascade));
+ } catch (SQLException e) {
+ if (DEPENDENT_OBJECTS_STILL_EXIST.equals(e.getSQLState())) {
+ throw new NonEmptySchemaException(
+ e,
+ "Cannot drop non-empty PostgreSQL schema %s: a schema-local
cascade cannot be "
+ + "guaranteed. Remove its objects explicitly, then retry
dropping the empty schema. "
+ + "PostgreSQL reason: %s",
+ schema,
+ e.getMessage());
+ }
+ throw exceptionMapper.toGravitinoException(e);
}
- return sqlBuilder.toString();
+ }
+
+ private static String quoteName(String name) {
+ return "\"" + name.replace("\"", "\"\"") + "\"";
}
@Override
diff --git
a/catalogs/catalog-jdbc-postgresql/src/test/java/org/apache/gravitino/catalog/postgresql/integration/test/CatalogPostgreSqlIT.java
b/catalogs/catalog-jdbc-postgresql/src/test/java/org/apache/gravitino/catalog/postgresql/integration/test/CatalogPostgreSqlIT.java
index e29616f4ec..6f78d3afa3 100644
---
a/catalogs/catalog-jdbc-postgresql/src/test/java/org/apache/gravitino/catalog/postgresql/integration/test/CatalogPostgreSqlIT.java
+++
b/catalogs/catalog-jdbc-postgresql/src/test/java/org/apache/gravitino/catalog/postgresql/integration/test/CatalogPostgreSqlIT.java
@@ -136,7 +136,7 @@ public class CatalogPostgreSqlIT extends BaseIT {
clearTableAndSchema();
String[] schemaNames = catalog.asSchemas().listSchemas();
for (String schemaName : schemaNames) {
- catalog.asSchemas().dropSchema(schemaName, true);
+ clearTableAndSchema(schemaName);
}
metalake.disableCatalog(catalogName);
metalake.dropCatalog(catalogName);
@@ -152,6 +152,12 @@ public class CatalogPostgreSqlIT extends BaseIT {
}
private void clearTableAndSchema() {
+ clearTableAndSchema(schemaName);
+ }
+
+ private void clearTableAndSchema(String schemaName) {
+ // PostgreSQL schema drops are restrictive even when cascade is requested.
Remove the test
+ // tables explicitly before dropping each schema, including schemas
created outside the fixture.
NameIdentifier[] nameIdentifiers =
catalog.asTableCatalog().listTables(Namespace.of(schemaName));
for (NameIdentifier nameIdentifier : nameIdentifiers) {
diff --git
a/catalogs/catalog-jdbc-postgresql/src/test/java/org/apache/gravitino/catalog/postgresql/operation/TestPostgreSqlSchemaOperations.java
b/catalogs/catalog-jdbc-postgresql/src/test/java/org/apache/gravitino/catalog/postgresql/operation/TestPostgreSqlSchemaOperations.java
index f59d7259a2..8750c85129 100644
---
a/catalogs/catalog-jdbc-postgresql/src/test/java/org/apache/gravitino/catalog/postgresql/operation/TestPostgreSqlSchemaOperations.java
+++
b/catalogs/catalog-jdbc-postgresql/src/test/java/org/apache/gravitino/catalog/postgresql/operation/TestPostgreSqlSchemaOperations.java
@@ -20,10 +20,14 @@ package org.apache.gravitino.catalog.postgresql.operation;
import com.google.common.collect.Maps;
import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
import java.sql.SQLException;
+import java.sql.Statement;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
import javax.sql.DataSource;
import org.apache.commons.lang3.StringUtils;
import org.apache.gravitino.catalog.jdbc.JdbcSchema;
@@ -31,6 +35,7 @@ import org.apache.gravitino.catalog.jdbc.config.JdbcConfig;
import org.apache.gravitino.catalog.jdbc.utils.DataSourceUtils;
import org.apache.gravitino.catalog.jdbc.utils.JdbcConnectorUtils;
import org.apache.gravitino.exceptions.NoSuchSchemaException;
+import org.apache.gravitino.exceptions.NonEmptySchemaException;
import org.apache.gravitino.utils.RandomNameUtils;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Tag;
@@ -39,6 +44,252 @@ import org.junit.jupiter.api.Test;
@Tag("gravitino-docker-test")
public class TestPostgreSqlSchemaOperations extends TestPostgreSql {
+ @Test
+ void testCascadeDropPreservesExternalDependents() throws SQLException {
+ String schema = RandomNameUtils.genRandomName("cascade_source");
+ String other = RandomNameUtils.genRandomName("cascade_other");
+ try (Connection connection = DATA_SOURCE.getConnection()) {
+ JdbcConnectorUtils.executeUpdate(connection, "CREATE SCHEMA " + schema);
+ JdbcConnectorUtils.executeUpdate(connection, "CREATE SCHEMA " + other);
+ try {
+ JdbcConnectorUtils.executeUpdate(
+ connection, "CREATE TABLE " + schema + ".parent (id int PRIMARY
KEY, payload text)");
+ JdbcConnectorUtils.executeUpdate(
+ connection, "CREATE VIEW " + other + ".v AS SELECT * FROM " +
schema + ".parent");
+ JdbcConnectorUtils.executeUpdate(
+ connection,
+ "CREATE TABLE " + other + ".child (id int REFERENCES " + schema +
".parent)");
+ NonEmptySchemaException error =
+ Assertions.assertThrows(
+ NonEmptySchemaException.class, () ->
DATABASE_OPERATIONS.delete(schema, true));
+ Assertions.assertTrue(error.getMessage().contains(schema));
+ Assertions.assertEquals("2BP01", ((SQLException)
error.getCause()).getSQLState());
+ try (Statement statement = connection.createStatement();
+ ResultSet rows = statement.executeQuery("SELECT * FROM " + other +
".v")) {
+ Assertions.assertFalse(rows.next());
+ }
+ Assertions.assertThrows(
+ SQLException.class,
+ () ->
+ JdbcConnectorUtils.executeUpdate(
+ connection, "INSERT INTO " + other + ".child VALUES
(123)"));
+ JdbcConnectorUtils.executeUpdate(connection, "DROP SCHEMA " + other +
" CASCADE");
+ Assertions.assertThrows(
+ NonEmptySchemaException.class, () ->
DATABASE_OPERATIONS.delete(schema, true));
+ JdbcConnectorUtils.executeUpdate(connection, "DROP TABLE " + schema +
".parent");
+ Assertions.assertTrue(DATABASE_OPERATIONS.delete(schema, true));
+ } finally {
+ JdbcConnectorUtils.executeUpdate(connection, "DROP SCHEMA IF EXISTS "
+ other + " CASCADE");
+ JdbcConnectorUtils.executeUpdate(
+ connection, "DROP SCHEMA IF EXISTS " + schema + " CASCADE");
+ }
+ }
+ }
+
+ @Test
+ void testCascadeDropPreservesExternalSequenceDefault() throws SQLException {
+ String schema = RandomNameUtils.genRandomName("cascade_sequence");
+ String other = RandomNameUtils.genRandomName("cascade_default");
+ try (Connection connection = DATA_SOURCE.getConnection()) {
+ JdbcConnectorUtils.executeUpdate(connection, "CREATE SCHEMA " + schema);
+ JdbcConnectorUtils.executeUpdate(connection, "CREATE SCHEMA " + other);
+ try {
+ JdbcConnectorUtils.executeUpdate(connection, "CREATE SEQUENCE " +
schema + ".seq");
+ JdbcConnectorUtils.executeUpdate(
+ connection,
+ "CREATE TABLE " + other + ".child (id bigint DEFAULT nextval('" +
schema + ".seq'))");
+ NonEmptySchemaException error =
+ Assertions.assertThrows(
+ NonEmptySchemaException.class, () ->
DATABASE_OPERATIONS.delete(schema, true));
+ Assertions.assertTrue(error.getMessage().contains(schema));
+ Assertions.assertEquals("2BP01", ((SQLException)
error.getCause()).getSQLState());
+ JdbcConnectorUtils.executeUpdate(
+ connection, "INSERT INTO " + other + ".child DEFAULT VALUES");
+ try (Statement statement = connection.createStatement();
+ ResultSet rows = statement.executeQuery("SELECT id FROM " + other
+ ".child")) {
+ Assertions.assertTrue(rows.next());
+ Assertions.assertEquals(1L, rows.getLong(1));
+ Assertions.assertFalse(rows.wasNull());
+ }
+ Assertions.assertEquals(schema,
DATABASE_OPERATIONS.load(schema).name());
+ } finally {
+ JdbcConnectorUtils.executeUpdate(connection, "DROP SCHEMA IF EXISTS "
+ other + " CASCADE");
+ JdbcConnectorUtils.executeUpdate(
+ connection, "DROP SCHEMA IF EXISTS " + schema + " CASCADE");
+ }
+ }
+ }
+
+ @Test
+ void testCascadeDropPreservesContainedToastObjects() throws SQLException {
+ String schema = RandomNameUtils.genRandomName("cascade_toast");
+ try (Connection connection = DATA_SOURCE.getConnection()) {
+ JdbcConnectorUtils.executeUpdate(connection, "CREATE SCHEMA " + schema);
+ try {
+ JdbcConnectorUtils.executeUpdate(
+ connection, "CREATE TABLE " + schema + ".payload (id int PRIMARY
KEY, body text)");
+ long toastId;
+ try (Statement statement = connection.createStatement();
+ ResultSet rows =
+ statement.executeQuery(
+ "SELECT reltoastrelid FROM pg_catalog.pg_class WHERE oid =
'"
+ + schema
+ + ".payload'::regclass")) {
+ Assertions.assertTrue(rows.next());
+ toastId = rows.getLong(1);
+ Assertions.assertTrue(toastId > 0, "The test table must have an
internal TOAST table");
+ }
+
+ Assertions.assertThrows(
+ NonEmptySchemaException.class, () ->
DATABASE_OPERATIONS.delete(schema, true));
+ Assertions.assertEquals(schema,
DATABASE_OPERATIONS.load(schema).name());
+ try (Statement statement = connection.createStatement();
+ ResultSet rows =
+ statement.executeQuery(
+ "SELECT COUNT(*) FROM pg_catalog.pg_class WHERE oid = " +
toastId)) {
+ Assertions.assertTrue(rows.next());
+ Assertions.assertEquals(1, rows.getInt(1));
+ }
+ } finally {
+ JdbcConnectorUtils.executeUpdate(
+ connection, "DROP SCHEMA IF EXISTS " + schema + " CASCADE");
+ }
+ }
+ }
+
+ @Test
+ void testSchemaOwnerDoesNotNeedTableLockPrivileges() throws SQLException {
+ String schema = RandomNameUtils.genRandomName("cascade_owner");
+ String owner = RandomNameUtils.genRandomName("schema_owner");
+ try (Connection connection = DATA_SOURCE.getConnection();
+ Connection ownerConnection =
+ DriverManager.getConnection(
+ CONTAINER.getJdbcUrl(), CONTAINER.getUsername(),
CONTAINER.getPassword())) {
+ JdbcConnectorUtils.executeUpdate(connection, "CREATE ROLE " + owner);
+ JdbcConnectorUtils.executeUpdate(
+ connection, "CREATE SCHEMA " + schema + " AUTHORIZATION " + owner);
+ JdbcConnectorUtils.executeUpdate(connection, "CREATE TABLE " + schema +
".t (id int)");
+ JdbcConnectorUtils.executeUpdate(ownerConnection, "SET ROLE " + owner);
+ PostgreSqlSchemaOperations operations =
+ new PostgreSqlSchemaOperations() {
+ @Override
+ protected Connection getConnection() throws SQLException {
+ Connection result =
+ DriverManager.getConnection(
+ CONTAINER.getJdbcUrl(), CONTAINER.getUsername(),
CONTAINER.getPassword());
+ JdbcConnectorUtils.executeUpdate(result, "SET ROLE " + owner);
+ return result;
+ }
+ };
+ operations.initialize(
+ DATA_SOURCE,
+ JDBC_EXCEPTION_CONVERTER,
+ Map.of(JdbcConfig.JDBC_DATABASE.getKey(),
CONTAINER.getDatabaseName()));
+ try {
+ NonEmptySchemaException error =
+ Assertions.assertThrows(
+ NonEmptySchemaException.class, () -> operations.delete(schema,
true));
+ Assertions.assertEquals("2BP01", ((SQLException)
error.getCause()).getSQLState());
+ JdbcConnectorUtils.executeUpdate(ownerConnection, "DROP TABLE " +
schema + ".t RESTRICT");
+ Assertions.assertTrue(operations.delete(schema, true));
+ Assertions.assertFalse(operations.delete(schema, true));
+ } finally {
+ JdbcConnectorUtils.executeUpdate(
+ connection, "DROP SCHEMA IF EXISTS " + schema + " CASCADE");
+ JdbcConnectorUtils.executeUpdate(connection, "DROP ROLE " + owner);
+ }
+ }
+ }
+
+ @Test
+ void testCascadeDropPreservesExternalAutomaticDependents() throws
SQLException {
+ String schema = RandomNameUtils.genRandomName("cascade_auto");
+ String other = RandomNameUtils.genRandomName("cascade_external");
+ try (Connection connection = DATA_SOURCE.getConnection()) {
+ JdbcConnectorUtils.executeUpdate(connection, "CREATE SCHEMA " + schema);
+ JdbcConnectorUtils.executeUpdate(connection, "CREATE SCHEMA " + other);
+ try {
+ JdbcConnectorUtils.executeUpdate(
+ connection,
+ "CREATE TABLE " + schema + ".parent (id int, v int) PARTITION BY
RANGE(id)");
+ JdbcConnectorUtils.executeUpdate(
+ connection,
+ "CREATE TABLE "
+ + other
+ + ".part PARTITION OF "
+ + schema
+ + ".parent FOR VALUES FROM (0) TO (100)");
+ JdbcConnectorUtils.executeUpdate(
+ connection, "CREATE STATISTICS " + other + ".st ON id, v FROM " +
schema + ".parent");
+ Assertions.assertThrows(
+ NonEmptySchemaException.class, () ->
DATABASE_OPERATIONS.delete(schema, true));
+ JdbcConnectorUtils.executeUpdate(connection, "INSERT INTO " + other +
".part VALUES (1,2)");
+ try (Statement statement = connection.createStatement();
+ ResultSet rows =
+ statement.executeQuery(
+ "SELECT COUNT(*) FROM pg_catalog.pg_statistic_ext WHERE
stxname = 'st' AND "
+ + "stxnamespace = '"
+ + other
+ + "'::regnamespace")) {
+ Assertions.assertTrue(rows.next());
+ Assertions.assertEquals(1, rows.getInt(1));
+ }
+ } finally {
+ JdbcConnectorUtils.executeUpdate(connection, "DROP SCHEMA IF EXISTS "
+ other + " CASCADE");
+ JdbcConnectorUtils.executeUpdate(
+ connection, "DROP SCHEMA IF EXISTS " + schema + " CASCADE");
+ }
+ }
+ }
+
+ @Test
+ void testDependencyCommittedImmediatelyBeforeDropIsPreserved() throws
SQLException {
+ String schema = RandomNameUtils.genRandomName("cascade_race");
+ String other = RandomNameUtils.genRandomName("cascade_race_other");
+ AtomicBoolean created = new AtomicBoolean();
+ try (Connection connection = DATA_SOURCE.getConnection()) {
+ JdbcConnectorUtils.executeUpdate(connection, "CREATE SCHEMA " + schema);
+ JdbcConnectorUtils.executeUpdate(connection, "CREATE SCHEMA " + other);
+ JdbcConnectorUtils.executeUpdate(
+ connection, "CREATE VIEW " + schema + ".v AS SELECT 1 AS id");
+ PostgreSqlSchemaOperations operations =
+ new PostgreSqlSchemaOperations() {
+ @Override
+ public String generateDropDatabaseSql(String name, boolean
cascade) {
+ try {
+ // A different connection commits the dependency after the old
preflight would
+ // have finished and immediately before the actual DROP
statement is generated.
+ JdbcConnectorUtils.executeUpdate(
+ connection, "CREATE VIEW " + other + ".v AS SELECT * FROM
" + schema + ".v");
+ created.set(true);
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+ return super.generateDropDatabaseSql(name, cascade);
+ }
+ };
+ operations.initialize(
+ DATA_SOURCE,
+ JDBC_EXCEPTION_CONVERTER,
+ Map.of(JdbcConfig.JDBC_DATABASE.getKey(),
CONTAINER.getDatabaseName()));
+ try {
+ Assertions.assertThrows(
+ NonEmptySchemaException.class, () -> operations.delete(schema,
true));
+ Assertions.assertTrue(created.get());
+ try (Statement statement = connection.createStatement();
+ ResultSet rows = statement.executeQuery("SELECT id FROM " + other
+ ".v")) {
+ Assertions.assertTrue(rows.next());
+ Assertions.assertEquals(1, rows.getInt(1));
+ }
+ Assertions.assertEquals(schema,
DATABASE_OPERATIONS.load(schema).name());
+ } finally {
+ JdbcConnectorUtils.executeUpdate(connection, "DROP SCHEMA IF EXISTS "
+ other + " CASCADE");
+ JdbcConnectorUtils.executeUpdate(
+ connection, "DROP SCHEMA IF EXISTS " + schema + " CASCADE");
+ }
+ }
+ }
+
@Test
void testQuotedSchemaNamesLoadWithoutExecutingSql() throws SQLException {
for (String schema : new String[] {"quoted';CREATE SCHEMA
injected_marker;--"}) {
diff --git
a/clients/client-java/src/main/java/org/apache/gravitino/client/BaseSchemaCatalog.java
b/clients/client-java/src/main/java/org/apache/gravitino/client/BaseSchemaCatalog.java
index 5fc840852f..530bad4a96 100644
---
a/clients/client-java/src/main/java/org/apache/gravitino/client/BaseSchemaCatalog.java
+++
b/clients/client-java/src/main/java/org/apache/gravitino/client/BaseSchemaCatalog.java
@@ -241,6 +241,8 @@ abstract class BaseSchemaCatalog extends CatalogDTO
* @param schemaName The name identifier of the schema.
* @return The {@link Schema} with specified identifier.
* @throws NoSuchSchemaException if the schema with specified identifier
does not exist.
+ * @throws SchemaAlreadyExistsException if an external schema carries
another existing schema's
+ * identifier. Remove the copied source marker before importing an
independent copy.
*/
@Override
public Schema loadSchema(String schemaName) throws NoSuchSchemaException {
@@ -290,10 +292,14 @@ abstract class BaseSchemaCatalog extends CatalogDTO
/**
* Drop the schema with specified identifier.
*
+ * <p>Cascade support depends on the catalog. PostgreSQL refuses non-empty
schemas even when
+ * cascade is requested, because native cascade may remove objects in other
schemas.
+ *
* @param schemaName The name identifier of the schema.
- * @param cascade Whether to drop all the tables under the schema.
+ * @param cascade Whether to request recursive deletion, subject to catalog
support.
* @return true if the schema is dropped successfully, false otherwise.
- * @throws NonEmptySchemaException if the schema is not empty and cascade is
false.
+ * @throws NonEmptySchemaException if the schema is non-empty and cascade is
false, or the catalog
+ * cannot safely cascade within the schema.
*/
@Override
public boolean dropSchema(String schemaName, boolean cascade) throws
NonEmptySchemaException {
diff --git a/clients/client-python/gravitino/api/supports_schemas.py
b/clients/client-python/gravitino/api/supports_schemas.py
index b81d74c153..b2825a0c73 100644
--- a/clients/client-python/gravitino/api/supports_schemas.py
+++ b/clients/client-python/gravitino/api/supports_schemas.py
@@ -75,6 +75,10 @@ class SupportsSchemas(ABC):
Returns:
True if the schema exists, false otherwise.
+
+ Raises:
+ SchemaAlreadyExistsException: If loading detects another existing
schema's identifier.
+ The conflict is propagated instead of returning False.
"""
try:
self.load_schema(schema_name)
@@ -118,6 +122,8 @@ class SupportsSchemas(ABC):
Raises:
NoSuchSchemaException: If the schema does not exist (optional).
+ SchemaAlreadyExistsException: If the source marker belongs to
another existing schema.
+ Remove the copied source marker before importing an
independent copy.
Returns:
A schema.
@@ -142,17 +148,21 @@ class SupportsSchemas(ABC):
@abstractmethod
def drop_schema(self, schema_name: str, cascade: bool) -> bool:
- """Drop a schema from the catalog. If cascade option is true,
recursively
- drop all objects within the schema.
+ """Drop a schema from the catalog.
+
+ Cascade support depends on the catalog. PostgreSQL refuses non-empty
+ schemas even when cascade is requested, because native cascade may
+ remove objects in other schemas.
Args:
schema_name: The name of the schema.
- cascade: If true, recursively drop all objects within the schema.
+ cascade: Whether to request recursive deletion, subject to catalog
support.
Returns:
True if the schema exists and is dropped successfully, false
otherwise.
Raises:
- NonEmptySchemaException: If the schema is not empty and cascade is
false.
+ NonEmptySchemaException: If the schema is non-empty and cascade is
false,
+ or the catalog cannot safely cascade within the schema.
"""
pass
diff --git a/clients/client-python/gravitino/client/base_schema_catalog.py
b/clients/client-python/gravitino/client/base_schema_catalog.py
index 39fe4fa1fe..9bc73213e0 100644
--- a/clients/client-python/gravitino/client/base_schema_catalog.py
+++ b/clients/client-python/gravitino/client/base_schema_catalog.py
@@ -236,6 +236,8 @@ class BaseSchemaCatalog(
Raises:
NoSuchSchemaException if the schema with specified identifier does
not exist.
+ SchemaAlreadyExistsException if the source marker belongs to
another existing schema.
+ Remove the copied source marker before importing an
independent copy.
Returns:
The Schema with specified identifier.
@@ -294,12 +296,17 @@ class BaseSchemaCatalog(
def drop_schema(self, schema_name: str, cascade: bool) -> bool:
"""Drop the schema with specified identifier.
+ Cascade support depends on the catalog. PostgreSQL refuses non-empty
+ schemas even when cascade is requested, because native cascade may
+ remove objects in other schemas.
+
Args:
schema_name: The name of the schema.
- cascade: Whether to drop all the tables under the schema.
+ cascade: Whether to request recursive deletion, subject to catalog
support.
Raises:
- NonEmptySchemaException if the schema is not empty and cascade is
false.
+ NonEmptySchemaException if the schema is non-empty and cascade is
false,
+ or the catalog cannot safely cascade within the schema.
Returns:
true if the schema is dropped successfully, false otherwise.
diff --git a/core/src/main/java/org/apache/gravitino/StringIdentifier.java
b/core/src/main/java/org/apache/gravitino/StringIdentifier.java
index 38c644944f..0fb7d4476a 100644
--- a/core/src/main/java/org/apache/gravitino/StringIdentifier.java
+++ b/core/src/main/java/org/apache/gravitino/StringIdentifier.java
@@ -178,7 +178,7 @@ public class StringIdentifier {
}
String pattern = "(" + STRING_COMMENT;
- int startIndex = comment.indexOf(pattern);
+ int startIndex = comment.lastIndexOf(pattern);
if (startIndex == -1) {
return null;
}
@@ -212,11 +212,11 @@ public class StringIdentifier {
StringIdentifier identifier = fromComment(comment);
if (identifier != null) {
String format = String.format(STRING_COMMENT_FORMAT, " ",
STRING_COMMENT, identifier);
- int indexOf = comment.indexOf(format);
+ int indexOf = comment.lastIndexOf(format);
if (indexOf == -1) {
format = String.format(STRING_COMMENT_FORMAT, "", STRING_COMMENT,
identifier);
- indexOf = comment.indexOf(format);
+ indexOf = comment.lastIndexOf(format);
}
if (indexOf != -1) {
diff --git
a/core/src/main/java/org/apache/gravitino/catalog/SchemaOperationDispatcher.java
b/core/src/main/java/org/apache/gravitino/catalog/SchemaOperationDispatcher.java
index 71428deb8a..421f0f416c 100644
---
a/core/src/main/java/org/apache/gravitino/catalog/SchemaOperationDispatcher.java
+++
b/core/src/main/java/org/apache/gravitino/catalog/SchemaOperationDispatcher.java
@@ -513,9 +513,11 @@ public class SchemaOperationDispatcher extends
OperationDispatcher implements Sc
* Drops a schema.
*
* @param ident The identifier of the schema to be dropped.
- * @param cascade If true, drops all tables within the schema as well.
+ * @param cascade Whether to request cascading deletion. A catalog may
reject a non-empty schema
+ * even when this is true if it cannot safely contain the deletion.
* @return True if the schema was successfully dropped, false if the schema
doesn't exist.
- * @throws NonEmptySchemaException If the schema contains tables and cascade
is set to false.
+ * @throws NonEmptySchemaException If the schema is non-empty and cascade is
false, or if the
+ * catalog cannot safely support the requested cascading deletion.
* @throws RuntimeException If an error occurs while dropping the schema.
*/
@Override
@@ -642,9 +644,8 @@ public class SchemaOperationDispatcher extends
OperationDispatcher implements Sc
// If the entity in the store doesn't match the one in the external
system, we use the data
// of external system to correct it.
LOG.warn(
- "The Schema uid {} existed but still needs to be imported, this
could be happened "
- + "when Schema is renamed by external systems not controlled by
Gravitino. In this case, "
- + "we need to overwrite the stored entity to keep consistency.",
+ "Verifying ownership of schema identifier {} before import; the
source may have been "
+ + "renamed externally or may carry a copied identifier.",
stringId);
uid = stringId.id();
} else {
@@ -652,6 +653,37 @@ public class SchemaOperationDispatcher extends
OperationDispatcher implements Sc
uid = idGenerator.nextId();
}
+ SchemaEntity previousEntity = null;
+ if (stringId != null) {
+ try {
+ for (SchemaEntity existing :
+ store.list(identifier.namespace(), SchemaEntity.class, SCHEMA)) {
+ if (existing.id() == uid) {
+ if (existing.nameIdentifier().equals(identifier)) {
+ previousEntity = existing;
+ break;
+ }
+ try {
+ doWithCatalog(
+ getCatalogIdentifier(existing.nameIdentifier()),
+ c -> c.doWithSchemaOps(ops ->
ops.loadSchema(existing.nameIdentifier())),
+ NoSuchSchemaException.class);
+ } catch (NoSuchSchemaException e) {
+ // Preserve a genuine external rename only after confirming the
old object is gone.
+ previousEntity = existing;
+ break;
+ }
+ throw new SchemaAlreadyExistsException(
+ "Schema %s carries the identifier of existing schema %s;
refusing to move its metadata",
+ identifier, existing.nameIdentifier());
+ }
+ }
+ } catch (IOException e) {
+ throw new RuntimeException(
+ "Failed to verify schema identifier ownership for " + identifier,
e);
+ }
+ }
+
SchemaEntity schemaEntity =
SchemaEntity.builder()
.withId(uid)
@@ -668,7 +700,21 @@ public class SchemaOperationDispatcher extends
OperationDispatcher implements Sc
.build())
.build();
try {
- store.put(schemaEntity, true);
+ if (previousEntity == null) {
+ // An untrusted source marker must never overwrite an entity in
another namespace.
+ store.put(schemaEntity, false);
+ } else {
+ store.update(
+ previousEntity.nameIdentifier(),
+ SchemaEntity.class,
+ SCHEMA,
+ existing -> {
+ if (!existing.id().equals(schemaEntity.id())) {
+ throw new OptimisticLockException("Schema identifier changed
during import");
+ }
+ return schemaEntity;
+ });
+ }
} catch (EntityAlreadyExistsException e) {
throw e;
} catch (OptimisticLockException e) {
diff --git
a/core/src/main/java/org/apache/gravitino/connector/SupportsSchemas.java
b/core/src/main/java/org/apache/gravitino/connector/SupportsSchemas.java
index 734c2e9508..c6dab7c858 100644
--- a/core/src/main/java/org/apache/gravitino/connector/SupportsSchemas.java
+++ b/core/src/main/java/org/apache/gravitino/connector/SupportsSchemas.java
@@ -59,6 +59,8 @@ public interface SupportsSchemas {
*
* @param ident The name identifier of the schema.
* @return True if the schema exists, false otherwise.
+ * @throws SchemaAlreadyExistsException If loading an external schema
detects an identifier owned
+ * by another existing schema; the conflict is propagated instead of
returning false.
*/
default boolean schemaExists(NameIdentifier ident) {
try {
@@ -88,6 +90,8 @@ public interface SupportsSchemas {
* @param ident The name identifier of the schema.
* @return A schema.
* @throws NoSuchSchemaException If the schema does not exist (optional).
+ * @throws SchemaAlreadyExistsException If an external schema carries the
identifier of another
+ * existing schema. Remove the copied source marker before importing an
independent copy.
*/
Schema loadSchema(NameIdentifier ident) throws NoSuchSchemaException;
@@ -102,17 +106,21 @@ public interface SupportsSchemas {
Schema alterSchema(NameIdentifier ident, SchemaChange... changes) throws
NoSuchSchemaException;
/**
- * Drop a schema from the catalog. If cascade option is true, recursively
drop all objects within
- * the schema.
+ * Drop a schema from the catalog. If cascade is true and supported by the
catalog, recursively
+ * drop all objects within the schema.
+ *
+ * <p>A catalog may refuse a non-empty schema even when cascade is requested
if it cannot safely
+ * restrict deletion to objects within that schema. PostgreSQL currently
refuses such requests.
*
* <p>If the catalog implementation does not support this operation, it may
throw {@link
* UnsupportedOperationException}.
*
* @param ident The name identifier of the schema.
- * @param cascade If true, recursively drop all objects within the schema.
+ * @param cascade Whether to request recursive deletion, subject to catalog
support.
* @return True if the schema exists and is dropped successfully, false if
the schema doesn't
* exist.
- * @throws NonEmptySchemaException If the schema is not empty and cascade is
false.
+ * @throws NonEmptySchemaException If the schema is non-empty and cascade is
false, or the catalog
+ * cannot safely cascade within the schema.
*/
boolean dropSchema(NameIdentifier ident, boolean cascade) throws
NonEmptySchemaException;
}
diff --git a/core/src/test/java/org/apache/gravitino/TestStringIdentifier.java
b/core/src/test/java/org/apache/gravitino/TestStringIdentifier.java
index 765eef77ce..c489e36765 100644
--- a/core/src/test/java/org/apache/gravitino/TestStringIdentifier.java
+++ b/core/src/test/java/org/apache/gravitino/TestStringIdentifier.java
@@ -25,6 +25,15 @@ import org.junit.jupiter.api.Test;
public class TestStringIdentifier {
+ @Test
+ void testAppendedMarkerWinsOverMarkerLikeUserComment() {
+ String userComment = "x (From Gravitino, DO NOT EDIT:
gravitino.v1.uid123)";
+ StringIdentifier ownId = StringIdentifier.fromId(456L);
+ String comment = StringIdentifier.addToComment(ownId, userComment);
+ Assertions.assertEquals(ownId.id(),
StringIdentifier.fromComment(comment).id());
+ Assertions.assertEquals(userComment,
StringIdentifier.removeIdFromComment(comment));
+ }
+
@Test
public void testCreateStringIdentifierFromId() {
long uid = 1L;
diff --git
a/core/src/test/java/org/apache/gravitino/catalog/TestSchemaOperationDispatcher.java
b/core/src/test/java/org/apache/gravitino/catalog/TestSchemaOperationDispatcher.java
index 9063fdb1cd..a8e54c0246 100644
---
a/core/src/test/java/org/apache/gravitino/catalog/TestSchemaOperationDispatcher.java
+++
b/core/src/test/java/org/apache/gravitino/catalog/TestSchemaOperationDispatcher.java
@@ -33,6 +33,7 @@ import com.google.common.collect.ImmutableMap;
import java.io.IOException;
import java.time.Instant;
import java.util.Arrays;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
@@ -47,6 +48,7 @@ import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.Namespace;
import org.apache.gravitino.Schema;
import org.apache.gravitino.SchemaChange;
+import org.apache.gravitino.StringIdentifier;
import org.apache.gravitino.auth.AuthConstants;
import org.apache.gravitino.connector.HiddenPropertyMaskUtils;
import org.apache.gravitino.connector.TestCatalogOperations;
@@ -83,6 +85,60 @@ public class TestSchemaOperationDispatcher extends
TestOperationDispatcher {
FieldUtils.writeField(GravitinoEnv.getInstance(), "lockManager", new
LockManager(config), true);
}
+ @Test
+ void testExternalRenamePreservesSchemaIdentity() throws Exception {
+ NameIdentifier original = NameIdentifier.of(metalake, catalog,
"external_rename_original");
+ NameIdentifier renamed = NameIdentifier.of(metalake, catalog,
"external_rename_new");
+ dispatcher.createSchema(original, "original", ImmutableMap.of("k1", "v1",
"k2", "v2"));
+ SchemaEntity originalEntity = entityStore.get(original, SCHEMA,
SchemaEntity.class);
+ Map<String, String> properties = new
HashMap<>(originalEntity.properties());
+ properties.put("k1", "v1");
+ properties.put("k2", "v2");
+ properties.put(ID_KEY,
StringIdentifier.fromId(originalEntity.id()).toString());
+ catalogManager.doWithCatalogWrapper(
+ NameIdentifier.of(metalake, catalog),
+ c -> {
+ c.doWithSchemaOps(
+ ops -> {
+ ops.dropSchema(original, false);
+ return ops.createSchema(renamed, "renamed", properties);
+ });
+ return null;
+ });
+ Assertions.assertEquals(renamed.name(),
dispatcher.loadSchema(renamed).name());
+ Assertions.assertEquals(
+ originalEntity.id(), entityStore.get(renamed, SCHEMA,
SchemaEntity.class).id());
+ Assertions.assertFalse(entityStore.exists(original, SCHEMA));
+ }
+
+ @Test
+ void testCopiedSchemaIdentifierCannotMoveExistingEntity() throws Exception {
+ NameIdentifier original = NameIdentifier.of(metalake, catalog,
"marker_original");
+ NameIdentifier copy = NameIdentifier.of(metalake, catalog, "marker_copy");
+ dispatcher.createSchema(original, "original", ImmutableMap.of("k1", "v1",
"k2", "v2"));
+ SchemaEntity originalEntity = entityStore.get(original, SCHEMA,
SchemaEntity.class);
+ Map<String, String> copiedProperties = new HashMap<>();
+ copiedProperties.put("k1", "v1");
+ copiedProperties.put("k2", "v2");
+ copiedProperties.put(ID_KEY,
StringIdentifier.fromId(originalEntity.id()).toString());
+ catalogManager.doWithCatalogWrapper(
+ NameIdentifier.of(metalake, catalog),
+ c -> {
+ c.doWithSchemaOps(ops -> ops.createSchema(copy, "copy",
copiedProperties));
+ return null;
+ });
+ Assertions.assertThrows(SchemaAlreadyExistsException.class, () ->
dispatcher.loadSchema(copy));
+ Assertions.assertEquals(
+ originalEntity.id(), entityStore.get(original, SCHEMA,
SchemaEntity.class).id());
+ Assertions.assertFalse(entityStore.exists(copy, SCHEMA));
+ Assertions.assertTrue(dispatcher.dropSchema(copy, false));
+ Assertions.assertEquals(
+ originalEntity.id(), entityStore.get(original, SCHEMA,
SchemaEntity.class).id());
+ Assertions.assertEquals(
+ originalEntity.properties(),
+ entityStore.get(original, SCHEMA, SchemaEntity.class).properties());
+ }
+
@Test
public void testCreateAndListSchemas() throws IOException {
Namespace ns = Namespace.of(metalake, catalog);
@@ -241,6 +297,9 @@ public class TestSchemaOperationDispatcher extends
TestOperationDispatcher {
// EntityAlreadyExistsException, and the dispatcher-level retry sees the
entity on the third
// get.
reset(entityStore);
+ doReturn(Collections.emptyList())
+ .when(entityStore)
+ .list(schemaIdent.namespace(), SchemaEntity.class, SCHEMA);
doThrow(new NoSuchEntityException("mock error"))
.doThrow(new NoSuchEntityException("mock error"))
.doReturn(concurrentSchemaEntity)
@@ -276,6 +335,9 @@ public class TestSchemaOperationDispatcher extends
TestOperationDispatcher {
// an entity with a mismatched ID (operateOnEntity returns null →
imported=false → error
// thrown).
reset(entityStore);
+ doReturn(Collections.emptyList())
+ .when(entityStore)
+ .list(schemaIdent.namespace(), SchemaEntity.class, SCHEMA);
doThrow(new NoSuchEntityException("mock error"))
.doThrow(new NoSuchEntityException("mock error"))
.doReturn(mismatchedSchemaEntity)
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestSchemaMetaService.java
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestSchemaMetaService.java
index 3f342ecf3e..7c51e764af 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestSchemaMetaService.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestSchemaMetaService.java
@@ -256,6 +256,20 @@ public class TestSchemaMetaService extends TestJDBCBackend
{
}
}
+ @TestTemplate
+ void testIdentifierCollisionCannotOverwriteExistingSchema() throws
IOException {
+ createAndInsertMakeLake(metalakeName);
+ createAndInsertCatalog(metalakeName, catalogName);
+ SchemaEntity original = createAndInsertSchema(metalakeName, catalogName,
"identity_original");
+ SchemaEntity copy =
+ createSchemaEntity(original.id(), original.namespace(),
"identity_copy", AUDIT_INFO);
+ assertThrows(EntityAlreadyExistsException.class, () ->
backend.insert(copy, false));
+ assertTrue(backend.exists(original.nameIdentifier(),
Entity.EntityType.SCHEMA));
+ assertFalse(backend.exists(copy.nameIdentifier(),
Entity.EntityType.SCHEMA));
+ SchemaEntity preserved = backend.get(original.nameIdentifier(),
Entity.EntityType.SCHEMA);
+ Assertions.assertEquals(original.id(), preserved.id());
+ }
+
@TestTemplate
public void testInsertAlreadyExistsException() throws IOException {
createAndInsertMakeLake(metalakeName);
diff --git a/docs/jdbc-postgresql-catalog.md b/docs/jdbc-postgresql-catalog.md
index 3bc301809f..6538e999b5 100644
--- a/docs/jdbc-postgresql-catalog.md
+++ b/docs/jdbc-postgresql-catalog.md
@@ -77,7 +77,7 @@ Sensitive catalog properties such as `jdbc-password` are
hidden from the default
- Gravitino schema corresponds to the PostgreSQL schema.
- Supports creating schema with comments.
- Supports dropping schema.
-- Supports cascade dropping schema.
+- Non-empty schema drops, including requests with `cascade=true`, are refused
with a schema-not-empty conflict (HTTP 409). PostgreSQL native `CASCADE` can
delete objects in other schemas, and a client-side dependency check cannot
prevent concurrent DDL from adding such dependencies. Remove the schema objects
explicitly, then drop the empty schema.
### Schema Properties
diff --git a/docs/open-api/schemas.yaml b/docs/open-api/schemas.yaml
index 3539e76da6..de5a3db0c6 100644
--- a/docs/open-api/schemas.yaml
+++ b/docs/open-api/schemas.yaml
@@ -96,6 +96,14 @@ paths:
$ref:
"./catalogs.yaml#/components/examples/NoSuchCatalogException"
NoSuchSchemaException:
$ref: "#/components/examples/NoSuchSchemaException"
+ "409":
+ description: >
+ Conflict - An external schema carries an identifier owned by
another existing schema.
+ Remove the copied source marker before importing an independent
copy.
+ content:
+ application/vnd.gravitino.v1+json:
+ schema:
+ $ref: "./openapi.yaml#/components/schemas/ErrorModel"
"5xx":
$ref: "./openapi.yaml#/components/responses/ServerErrorResponse"
@@ -148,6 +156,14 @@ paths:
$ref: "./openapi.yaml#/components/responses/DropResponse"
"400":
$ref: "./openapi.yaml#/components/responses/BadRequestErrorResponse"
+ "409":
+ description: >
+ Conflict - The schema is not empty, or the catalog cannot safely
support the requested
+ cascading deletion. PostgreSQL rejects non-empty schema deletion
even with cascade.
+ content:
+ application/vnd.gravitino.v1+json:
+ schema:
+ $ref: "./openapi.yaml#/components/schemas/ErrorModel"
"5xx":
$ref: "./openapi.yaml#/components/responses/ServerErrorResponse"
diff --git
a/trino-connector/integration-test/src/test/resources/trino-ci-testset/testsets/jdbc-postgresql/00006_datatype.sql
b/trino-connector/integration-test/src/test/resources/trino-ci-testset/testsets/jdbc-postgresql/00006_datatype.sql
index 57fd701349..29b4edeae7 100644
---
a/trino-connector/integration-test/src/test/resources/trino-ci-testset/testsets/jdbc-postgresql/00006_datatype.sql
+++
b/trino-connector/integration-test/src/test/resources/trino-ci-testset/testsets/jdbc-postgresql/00006_datatype.sql
@@ -71,6 +71,9 @@ INSERT INTO tb02 (f1, f2, f3, f4, f5, f6, f7, f9, f10, f11,
f12, f13, f14, f15,
VALUES ('Sample text 1', 'same9', x'65', 123.456, 7.89, 12.34, false, 100,
1000, 1000, 1992382342, DATE '2024-01-01',
NULL, TIMESTAMP '2024-01-01 08:00:00', TIMESTAMP '2024-01-01 08:00:00
UTC');
+-- Failed inserts must roll back their JDBC staging tables.
+SHOW TABLES;
+
CREATE TABLE tb03 (int_array ARRAY(INTEGER));
INSERT INTO tb03 VALUES (ARRAY[1, 2, 3, 4, 5]);
@@ -83,4 +86,4 @@ drop table tb02;
drop table tb03;
-drop schema gt_postgresql.gt_db1 cascade;
+drop schema gt_postgresql.gt_db1;
diff --git
a/trino-connector/integration-test/src/test/resources/trino-ci-testset/testsets/jdbc-postgresql/00006_datatype.txt
b/trino-connector/integration-test/src/test/resources/trino-ci-testset/testsets/jdbc-postgresql/00006_datatype.txt
index e37cbee1a5..09f6f4be73 100644
---
a/trino-connector/integration-test/src/test/resources/trino-ci-testset/testsets/jdbc-postgresql/00006_datatype.txt
+++
b/trino-connector/integration-test/src/test/resources/trino-ci-testset/testsets/jdbc-postgresql/00006_datatype.txt
@@ -61,6 +61,9 @@ INSERT: 1 row
<QUERY_FAILED> NULL value not allowed for NOT NULL column: f14
+"tb01"
+"tb02"
+
CREATE TABLE
INSERT: 1 row
diff --git
a/trino-connector/src/common/main/java/org/apache/gravitino/trino/connector/GravitinoConnector.java
b/trino-connector/src/common/main/java/org/apache/gravitino/trino/connector/GravitinoConnector.java
index 87f8fe9e5c..188c9cac50 100644
---
a/trino-connector/src/common/main/java/org/apache/gravitino/trino/connector/GravitinoConnector.java
+++
b/trino-connector/src/common/main/java/org/apache/gravitino/trino/connector/GravitinoConnector.java
@@ -232,6 +232,14 @@ public class GravitinoConnector implements Connector {
internalConnector.commit(gravitinoTransactionHandle.getInternalHandle());
}
+ @Override
+ public void rollback(ConnectorTransactionHandle transactionHandle) {
+ GravitinoTransactionHandle gravitinoTransactionHandle =
+ (GravitinoTransactionHandle) transactionHandle;
+ Connector internalConnector =
catalogConnectorContext.getInternalConnector();
+ internalConnector.rollback(gravitinoTransactionHandle.getInternalHandle());
+ }
+
@Override
public ConnectorAccessControl getAccessControl() {
Connector internalConnector =
catalogConnectorContext.getInternalConnector();
diff --git
a/trino-connector/src/common/test/java/org/apache/gravitino/trino/connector/TestGravitinoConnectorTransactions.java
b/trino-connector/src/common/test/java/org/apache/gravitino/trino/connector/TestGravitinoConnectorTransactions.java
new file mode 100644
index 0000000000..cd72fa4ca5
--- /dev/null
+++
b/trino-connector/src/common/test/java/org/apache/gravitino/trino/connector/TestGravitinoConnectorTransactions.java
@@ -0,0 +1,100 @@
+/*
+ * 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 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.trino.connector;
+
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.when;
+
+import com.google.common.collect.ImmutableMap;
+import io.trino.spi.connector.Connector;
+import io.trino.spi.connector.ConnectorTransactionHandle;
+import org.apache.gravitino.Catalog;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.SupportsSchemas;
+import org.apache.gravitino.client.GravitinoMetalake;
+import org.apache.gravitino.rel.TableCatalog;
+import org.apache.gravitino.trino.connector.catalog.CatalogConnectorContext;
+import org.apache.gravitino.trino.connector.metadata.GravitinoCatalog;
+import org.junit.jupiter.api.Test;
+
+class TestGravitinoConnectorTransactions {
+ @Test
+ void testRollbackDelegatesToInternalTransaction() {
+ Connector delegate = mock(Connector.class);
+ GravitinoConnector connector = connector(delegate);
+ ConnectorTransactionHandle transaction =
mock(ConnectorTransactionHandle.class);
+
+ connector.rollback(new GravitinoTransactionHandle(transaction));
+
+ verify(delegate).rollback(transaction);
+ verifyNoMoreInteractions(delegate);
+ }
+
+ @Test
+ void testRollbackPropagatesCleanupFailure() {
+ Connector delegate = mock(Connector.class);
+ GravitinoConnector connector = connector(delegate);
+ ConnectorTransactionHandle transaction =
mock(ConnectorTransactionHandle.class);
+ RuntimeException failure = new IllegalStateException("Failed to remove
staging table");
+ doThrow(failure).when(delegate).rollback(transaction);
+
+ assertSame(
+ failure,
+ assertThrows(
+ IllegalStateException.class,
+ () -> connector.rollback(new
GravitinoTransactionHandle(transaction))));
+ verify(delegate).rollback(transaction);
+ verifyNoMoreInteractions(delegate);
+ }
+
+ @Test
+ void testCommitDelegatesToInternalTransaction() {
+ Connector delegate = mock(Connector.class);
+ GravitinoConnector connector = connector(delegate);
+ ConnectorTransactionHandle transaction =
mock(ConnectorTransactionHandle.class);
+
+ connector.commit(new GravitinoTransactionHandle(transaction));
+
+ verify(delegate).commit(transaction);
+ verifyNoMoreInteractions(delegate);
+ }
+
+ private GravitinoConnector connector(Connector delegate) {
+ GravitinoCatalog catalog = mock(GravitinoCatalog.class);
+ when(catalog.geNameIdentifier()).thenReturn(NameIdentifier.of("metalake",
"catalog"));
+ CatalogConnectorContext context = mock(CatalogConnectorContext.class);
+ when(context.getCatalog()).thenReturn(catalog);
+ when(context.getInternalConnector()).thenReturn(delegate);
+ when(context.getConfig())
+ .thenReturn(new GravitinoConfig(ImmutableMap.of("gravitino.metalake",
"metalake")));
+ GravitinoMetalake metalake = mock(GravitinoMetalake.class);
+ Catalog clientCatalog = mock(Catalog.class);
+ when(clientCatalog.asSchemas()).thenReturn(mock(SupportsSchemas.class));
+ when(clientCatalog.asTableCatalog()).thenReturn(mock(TableCatalog.class));
+ when(metalake.loadCatalog(any())).thenReturn(clientCatalog);
+ when(context.getMetalake()).thenReturn(metalake);
+ return new GravitinoConnector(context);
+ }
+}
