This is an automated email from the ASF dual-hosted git repository. jerryshao pushed a commit to branch branch-1.3 in repository https://gitbox.apache.org/repos/asf/gravitino.git
commit 8055d333d9e1e7786773633a9ecf1705c8e3c397 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) 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. 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. 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. 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. (cherry picked from commit 27201d342e6e289db00dda8a639da01c1f49d82b) --- .../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 | 64 +++++- .../gravitino/connector/SupportsSchemas.java | 16 +- .../org/apache/gravitino/TestStringIdentifier.java | 9 + .../catalog/TestSchemaOperationDispatcher.java | 92 ++++++++ .../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, 667 insertions(+), 34 deletions(-) diff --git a/api/src/main/java/org/apache/gravitino/SupportsSchemas.java b/api/src/main/java/org/apache/gravitino/SupportsSchemas.java index 42284ed24d..365272e40a 100644 --- a/api/src/main/java/org/apache/gravitino/SupportsSchemas.java +++ b/api/src/main/java/org/apache/gravitino/SupportsSchemas.java @@ -76,6 +76,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 { @@ -109,6 +111,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; @@ -123,17 +127,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 414d198db4..dd80fd5b62 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 @@ -135,7 +135,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); @@ -151,6 +151,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..9b0fa81f7c 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; @@ -53,6 +58,48 @@ 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 testBackslashSchemaNameLoads() { String schema = "back\\slash_%"; @@ -65,6 +112,210 @@ public class TestPostgreSqlSchemaOperations extends TestPostgreSql { } } + @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 public void testBaseOperationSchema() { String databaseName = RandomNameUtils.genRandomName("ct_db"); 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 c8a3d2be52..8ff82d5bd7 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 @@ -203,6 +203,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 { @@ -252,10 +254,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 7a61400b6f..d1054799e0 100644 --- a/clients/client-python/gravitino/api/supports_schemas.py +++ b/clients/client-python/gravitino/api/supports_schemas.py @@ -70,6 +70,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) @@ -106,6 +110,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. @@ -130,17 +136,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 64f9c482d0..7eb4c510b6 100644 --- a/clients/client-python/gravitino/client/base_schema_catalog.py +++ b/clients/client-python/gravitino/client/base_schema_catalog.py @@ -201,6 +201,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. @@ -259,12 +261,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 54ea2df613..ae5500fb3c 100644 --- a/core/src/main/java/org/apache/gravitino/catalog/SchemaOperationDispatcher.java +++ b/core/src/main/java/org/apache/gravitino/catalog/SchemaOperationDispatcher.java @@ -22,6 +22,7 @@ import static org.apache.gravitino.Entity.EntityType.SCHEMA; import static org.apache.gravitino.catalog.PropertiesMetadataHelpers.validatePropertyForCreate; import static org.apache.gravitino.utils.NameIdentifierUtil.getCatalogIdentifier; +import java.io.IOException; import java.time.Instant; import java.util.Map; import org.apache.gravitino.EntityAlreadyExistsException; @@ -314,9 +315,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 @@ -383,9 +386,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 { @@ -393,6 +395,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) @@ -407,8 +440,25 @@ public class SchemaOperationDispatcher extends OperationDispatcher implements Sc .build()) .build(); try { - store.put(schemaEntity, true); - } catch (EntityAlreadyExistsException e) { + 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())) { + // branch-1.3 uses the existing schema-conflict response for import races. + throw new SchemaAlreadyExistsException( + "Schema identifier changed while importing %s; retry loading the schema", + identifier); + } + return schemaEntity; + }); + } + } catch (EntityAlreadyExistsException | SchemaAlreadyExistsException e) { throw e; } catch (Exception e) { LOG.error(FormattedErrorMessages.STORE_OP_FAILURE, "put", identifier, 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 c452699214..c4d0069706 100644 --- a/core/src/test/java/org/apache/gravitino/catalog/TestSchemaOperationDispatcher.java +++ b/core/src/test/java/org/apache/gravitino/catalog/TestSchemaOperationDispatcher.java @@ -21,9 +21,11 @@ package org.apache.gravitino.catalog; import static org.apache.gravitino.Entity.EntityType.SCHEMA; import static org.apache.gravitino.StringIdentifier.ID_KEY; import static org.apache.gravitino.TestBasePropertiesMetadata.COMMENT_KEY; +import static org.mockito.AdditionalAnswers.delegatesTo; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyBoolean; import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; @@ -33,21 +35,26 @@ 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; +import java.util.function.Function; import org.apache.commons.lang3.reflect.FieldUtils; import org.apache.gravitino.Config; import org.apache.gravitino.Configs; import org.apache.gravitino.Entity; import org.apache.gravitino.EntityAlreadyExistsException; +import org.apache.gravitino.EntityStore; import org.apache.gravitino.GravitinoEnv; 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.exceptions.NoSuchEntityException; +import org.apache.gravitino.exceptions.SchemaAlreadyExistsException; import org.apache.gravitino.lock.LockManager; import org.apache.gravitino.meta.AuditInfo; import org.apache.gravitino.meta.SchemaEntity; @@ -70,6 +77,85 @@ 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<>(); + 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; + }); + EntityStore conflictingStore = mock(EntityStore.class, delegatesTo(entityStore)); + doAnswer( + invocation -> { + Function<SchemaEntity, SchemaEntity> updater = invocation.getArgument(3); + return updater.apply( + SchemaEntity.builder() + .withId(originalEntity.id() + 1) + .withName(original.name()) + .withNamespace(original.namespace()) + .withAuditInfo(originalEntity.auditInfo()) + .build()); + }) + .when(conflictingStore) + .update(eq(original), eq(SchemaEntity.class), eq(SCHEMA), any()); + SchemaOperationDispatcher conflictingDispatcher = + new SchemaOperationDispatcher(catalogManager, conflictingStore, idGenerator); + SchemaAlreadyExistsException conflict = + Assertions.assertThrows( + SchemaAlreadyExistsException.class, () -> conflictingDispatcher.loadSchema(renamed)); + Assertions.assertTrue(conflict.getMessage().contains(renamed.toString())); + Assertions.assertTrue(conflict.getMessage().contains("retry")); + Assertions.assertEquals( + originalEntity.id(), entityStore.get(original, SCHEMA, SchemaEntity.class).id()); + Assertions.assertFalse(entityStore.exists(renamed, SCHEMA)); + + 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); @@ -228,6 +314,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) @@ -263,6 +352,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 3b4241a99a..839c8a3507 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 @@ -243,6 +243,20 @@ public class TestSchemaMetaService extends TestJDBCBackend { assertTrue(backend.exists(function.nameIdentifier(), Entity.EntityType.FUNCTION)); } + @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 b5def139a8..f545d6740a 100644 --- a/docs/jdbc-postgresql-catalog.md +++ b/docs/jdbc-postgresql-catalog.md @@ -73,7 +73,7 @@ Sensitive catalog properties such as `jdbc-user` and `jdbc-password` are hidden - 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 06f90f5056..8c660fbdcb 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/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoConnector.java b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoConnector.java index 8f130f2974..cfa4d6722b 100644 --- a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoConnector.java +++ b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoConnector.java @@ -226,6 +226,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/trino-connector/src/test/java/org/apache/gravitino/trino/connector/TestGravitinoConnectorTransactions.java b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/TestGravitinoConnectorTransactions.java new file mode 100644 index 0000000000..cd72fa4ca5 --- /dev/null +++ b/trino-connector/trino-connector/src/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); + } +}
