diff --git a/core/src/main/java/org/apache/gravitino/metalake/MetalakeManager.java b/core/src/main/java/org/apache/gravitino/metalake/MetalakeManager.java index 1bc5dfd13bd..8f4a038ae4a 100644 --- a/core/src/main/java/org/apache/gravitino/metalake/MetalakeManager.java +++ b/core/src/main/java/org/apache/gravitino/metalake/MetalakeManager.java @@ -358,7 +358,9 @@ public boolean dropMetalake(NameIdentifier ident, boolean force) } return store.delete(ident, EntityType.METALAKE, true); - } catch (NoSuchMetalakeException e) { + } catch (NoSuchMetalakeException | NoSuchEntityException e) { + // Another server may have completed the drop after the initial existence check. + // Dropping an already-removed metalake remains an idempotent false result. return false; } catch (IOException e) { diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/CatalogMetaMapper.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/CatalogMetaMapper.java index 9f19d1a6a84..553a33a71a9 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/CatalogMetaMapper.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/CatalogMetaMapper.java @@ -47,6 +47,12 @@ public interface CatalogMetaMapper { @SelectProvider(type = CatalogMetaSQLProviderFactory.class, method = "listCatalogPOsByMetalakeId") List listCatalogPOsByMetalakeId(@Param("metalakeId") Long metalakeId); + /** Selects and locks all active catalogs in a metalake for the current transaction. */ + @SelectProvider( + type = CatalogMetaSQLProviderFactory.class, + method = "listCatalogPOsByMetalakeIdForUpdate") + List listCatalogPOsByMetalakeIdForUpdate(@Param("metalakeId") Long metalakeId); + @SelectProvider(type = CatalogMetaSQLProviderFactory.class, method = "listCatalogPOsByCatalogIds") List listCatalogPOsByCatalogIds(@Param("catalogIds") List catalogIds); @@ -91,10 +97,15 @@ Integer updateCatalogMeta( method = "softDeleteCatalogMetasByCatalogId") Integer softDeleteCatalogMetasByCatalogId(@Param("catalogId") Long catalogId); + /** + * Soft-deletes catalogs whose identifiers and OCC versions still match. + * + * @return the number of deleted rows + */ @UpdateProvider( type = CatalogMetaSQLProviderFactory.class, - method = "softDeleteCatalogMetasByMetalakeId") - Integer softDeleteCatalogMetasByMetalakeId(@Param("metalakeId") Long metalakeId); + method = "softDeleteCatalogMetasWithVersion") + Integer softDeleteCatalogMetasWithVersion(@Param("catalogMetas") List catalogPOs); @DeleteProvider( type = CatalogMetaSQLProviderFactory.class, diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/CatalogMetaSQLProviderFactory.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/CatalogMetaSQLProviderFactory.java index c3a7954a25a..9b3154fa9ec 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/CatalogMetaSQLProviderFactory.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/CatalogMetaSQLProviderFactory.java @@ -61,6 +61,11 @@ public static String listCatalogPOsByMetalakeId(@Param("metalakeId") Long metala return getProvider().listCatalogPOsByMetalakeId(metalakeId); } + /** Returns SQL that lists and locks all active catalogs in a metalake. */ + public static String listCatalogPOsByMetalakeIdForUpdate(@Param("metalakeId") Long metalakeId) { + return getProvider().listCatalogPOsByMetalakeIdForUpdate(metalakeId); + } + public static String listCatalogPOsByCatalogIds(@Param("catalogIds") List catalogIds) { return getProvider().listCatalogPOsByCatalogIds(catalogIds); } @@ -113,8 +118,10 @@ public static String softDeleteCatalogMetasByCatalogId(@Param("catalogId") Long return getProvider().softDeleteCatalogMetasByCatalogId(catalogId); } - public static String softDeleteCatalogMetasByMetalakeId(@Param("metalakeId") Long metalakeId) { - return getProvider().softDeleteCatalogMetasByMetalakeId(metalakeId); + /** Returns SQL that soft-deletes catalogs using identifier-and-version pairs. */ + public static String softDeleteCatalogMetasWithVersion( + @Param("catalogMetas") List catalogPOs) { + return getProvider().softDeleteCatalogMetasWithVersion(catalogPOs); } public static String deleteCatalogMetasByLegacyTimeline( diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/MetalakeMetaMapper.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/MetalakeMetaMapper.java index f705c283ce6..665ae02b794 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/MetalakeMetaMapper.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/MetalakeMetaMapper.java @@ -47,6 +47,12 @@ public interface MetalakeMetaMapper { @SelectProvider(type = MetalakeMetaSQLProviderFactory.class, method = "selectMetalakeMetaById") MetalakePO selectMetalakeMetaById(@Param("metalakeId") Long metalakeId); + /** Selects and locks an active metalake by ID for the current transaction. */ + @SelectProvider( + type = MetalakeMetaSQLProviderFactory.class, + method = "selectMetalakeMetaByIdForUpdate") + MetalakePO selectMetalakeMetaByIdForUpdate(@Param("metalakeId") Long metalakeId); + @SelectProvider( type = MetalakeMetaSQLProviderFactory.class, method = "listMetalakePOsByMetalakeIds") @@ -65,15 +71,26 @@ public interface MetalakeMetaMapper { method = "insertMetalakeMetaOnDuplicateKeyUpdate") void insertMetalakeMetaOnDuplicateKeyUpdate(@Param("metalakeMeta") MetalakePO metalakePO); + /** + * Updates an active metalake only when its OCC version still matches the observed version. + * + * @return the number of updated rows + */ @UpdateProvider(type = MetalakeMetaSQLProviderFactory.class, method = "updateMetalakeMeta") Integer updateMetalakeMeta( @Param("newMetalakeMeta") MetalakePO newMetalakePO, @Param("oldMetalakeMeta") MetalakePO oldMetalakePO); + /** + * Soft-deletes an active metalake only when its OCC version still matches. + * + * @return the number of deleted rows + */ @UpdateProvider( type = MetalakeMetaSQLProviderFactory.class, method = "softDeleteMetalakeMetaByMetalakeId") - Integer softDeleteMetalakeMetaByMetalakeId(@Param("metalakeId") Long metalakeId); + Integer softDeleteMetalakeMetaByMetalakeId( + @Param("metalakeId") Long metalakeId, @Param("currentVersion") Long currentVersion); @DeleteProvider( type = MetalakeMetaSQLProviderFactory.class, diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/MetalakeMetaSQLProviderFactory.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/MetalakeMetaSQLProviderFactory.java index eba26f9e025..11f64ad662b 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/MetalakeMetaSQLProviderFactory.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/MetalakeMetaSQLProviderFactory.java @@ -65,6 +65,11 @@ public static String selectMetalakeMetaById(@Param("metalakeId") Long metalakeId return getProvider().selectMetalakeMetaById(metalakeId); } + /** Returns SQL that selects and locks an active metalake by ID. */ + public static String selectMetalakeMetaByIdForUpdate(@Param("metalakeId") Long metalakeId) { + return getProvider().selectMetalakeMetaByIdForUpdate(metalakeId); + } + public static String selectMetalakeIdMetaByName(@Param("metalakeName") String metalakeName) { return getProvider().selectMetalakeIdMetaByName(metalakeName); } @@ -88,8 +93,9 @@ public static String updateMetalakeMeta( return getProvider().updateMetalakeMeta(newMetalakePO, oldMetalakePO); } - public static String softDeleteMetalakeMetaByMetalakeId(@Param("metalakeId") Long metalakeId) { - return getProvider().softDeleteMetalakeMetaByMetalakeId(metalakeId); + public static String softDeleteMetalakeMetaByMetalakeId( + @Param("metalakeId") Long metalakeId, @Param("currentVersion") Long currentVersion) { + return getProvider().softDeleteMetalakeMetaByMetalakeId(metalakeId, currentVersion); } public static String deleteMetalakeMetasByLegacyTimeline( diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/SchemaMetaMapper.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/SchemaMetaMapper.java index 1c9b5286b29..989f1ebd99e 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/SchemaMetaMapper.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/SchemaMetaMapper.java @@ -42,6 +42,10 @@ public interface SchemaMetaMapper { @SelectProvider(type = SchemaMetaSQLProviderFactory.class, method = "listSchemaPOsByCatalogId") List listSchemaPOsByCatalogId(@Param("catalogId") Long catalogId); + /** Lists all active schemas in a metalake. */ + @SelectProvider(type = SchemaMetaSQLProviderFactory.class, method = "listSchemaPOsByMetalakeId") + List listSchemaPOsByMetalakeId(@Param("metalakeId") Long metalakeId); + @SelectProvider( type = SchemaMetaSQLProviderFactory.class, method = "listSchemaPOsByFullQualifiedName") @@ -109,13 +113,18 @@ Integer updateSchemaMeta( @UpdateProvider( type = SchemaMetaSQLProviderFactory.class, - method = "softDeleteSchemaMetasByMetalakeId") - Integer softDeleteSchemaMetasByMetalakeId(@Param("metalakeId") Long metalakeId); + method = "softDeleteSchemaMetasByCatalogId") + Integer softDeleteSchemaMetasByCatalogId(@Param("catalogId") Long catalogId); + /** + * Soft-deletes schemas whose identifiers and OCC versions still match. + * + * @return the number of deleted rows + */ @UpdateProvider( type = SchemaMetaSQLProviderFactory.class, - method = "softDeleteSchemaMetasByCatalogId") - Integer softDeleteSchemaMetasByCatalogId(@Param("catalogId") Long catalogId); + method = "softDeleteSchemaMetasWithVersion") + Integer softDeleteSchemaMetasWithVersion(@Param("schemaMetas") List schemaPOs); @DeleteProvider( type = SchemaMetaSQLProviderFactory.class, diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/SchemaMetaSQLProviderFactory.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/SchemaMetaSQLProviderFactory.java index acc27170269..30c2ee9dfc2 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/SchemaMetaSQLProviderFactory.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/SchemaMetaSQLProviderFactory.java @@ -72,6 +72,11 @@ public static String listSchemaPOsByCatalogId(@Param("catalogId") Long catalogId return getProvider().listSchemaPOsByCatalogId(catalogId); } + /** Returns SQL that lists all active schemas in a metalake. */ + public static String listSchemaPOsByMetalakeId(@Param("metalakeId") Long metalakeId) { + return getProvider().listSchemaPOsByMetalakeId(metalakeId); + } + public static String selectSchemaIdByCatalogIdAndName( @Param("catalogId") Long catalogId, @Param("schemaName") String name) { return getProvider().selectSchemaIdByCatalogIdAndName(catalogId, name); @@ -120,14 +125,16 @@ public static String softDeleteSchemaMetasBySchemaIds(@Param("schemaIds") List schemaPOs) { + return getProvider().softDeleteSchemaMetasWithVersion(schemaPOs); + } + public static String deleteSchemaMetasByLegacyTimeline( @Param("legacyTimeline") Long legacyTimeline, @Param("limit") int limit) { return getProvider().deleteSchemaMetasByLegacyTimeline(legacyTimeline, limit); diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/CatalogMetaBaseSQLProvider.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/CatalogMetaBaseSQLProvider.java index be03900dc22..f50b9c203d6 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/CatalogMetaBaseSQLProvider.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/CatalogMetaBaseSQLProvider.java @@ -53,6 +53,11 @@ public String listCatalogPOsByMetalakeId(@Param("metalakeId") Long metalakeId) { + " WHERE metalake_id = #{metalakeId} AND deleted_at = 0"; } + /** Returns SQL that lists and locks all active catalogs in a metalake. */ + public String listCatalogPOsByMetalakeIdForUpdate(@Param("metalakeId") Long metalakeId) { + return listCatalogPOsByMetalakeId(metalakeId) + " FOR UPDATE"; + } + public String listCatalogPOsByCatalogIds(@Param("catalogIds") List catalogIds) { return ""; } public String deleteCatalogMetasByLegacyTimeline( diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/MetalakeMetaBaseSQLProvider.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/MetalakeMetaBaseSQLProvider.java index 2524eda76fc..a7a78f4b7b5 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/MetalakeMetaBaseSQLProvider.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/MetalakeMetaBaseSQLProvider.java @@ -59,6 +59,11 @@ public String selectMetalakeMetaById(@Param("metalakeId") Long metalakeId) { + " WHERE metalake_id = #{metalakeId} AND deleted_at = 0"; } + /** Returns SQL that selects and locks an active metalake by ID. */ + public String selectMetalakeMetaByIdForUpdate(@Param("metalakeId") Long metalakeId) { + return selectMetalakeMetaById(metalakeId) + " FOR UPDATE"; + } + public String selectMetalakeIdMetaByName(@Param("metalakeName") String metalakeName) { return "SELECT metalake_id as metalakeId" + " FROM " @@ -143,23 +148,18 @@ public String updateMetalakeMeta( + " current_version = #{newMetalakeMeta.currentVersion}," + " last_version = #{newMetalakeMeta.lastVersion}" + " WHERE metalake_id = #{oldMetalakeMeta.metalakeId}" - + " AND metalake_name = #{oldMetalakeMeta.metalakeName}" - + " AND (metalake_comment = #{oldMetalakeMeta.metalakeComment} " - + " OR (metalake_comment IS NULL and #{oldMetalakeMeta.metalakeComment} IS NULL))" - + " AND properties = #{oldMetalakeMeta.properties}" - + " AND audit_info = #{oldMetalakeMeta.auditInfo}" - + " AND schema_version = #{oldMetalakeMeta.schemaVersion}" + " AND current_version = #{oldMetalakeMeta.currentVersion}" - + " AND last_version = #{oldMetalakeMeta.lastVersion}" + " AND deleted_at = 0"; } - public String softDeleteMetalakeMetaByMetalakeId(@Param("metalakeId") Long metalakeId) { + public String softDeleteMetalakeMetaByMetalakeId( + @Param("metalakeId") Long metalakeId, @Param("currentVersion") Long currentVersion) { return "UPDATE " + TABLE_NAME + " SET deleted_at = (UNIX_TIMESTAMP() * 1000.0)" + " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000" - + " WHERE metalake_id = #{metalakeId} AND deleted_at = 0"; + + " WHERE metalake_id = #{metalakeId}" + + " AND current_version = #{currentVersion} AND deleted_at = 0"; } public String deleteMetalakeMetasByLegacyTimeline( diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/SchemaMetaBaseSQLProvider.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/SchemaMetaBaseSQLProvider.java index 822ac3cf259..ee72e383249 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/SchemaMetaBaseSQLProvider.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/SchemaMetaBaseSQLProvider.java @@ -38,6 +38,18 @@ public String listSchemaPOsByCatalogId(@Param("catalogId") Long catalogId) { + " WHERE catalog_id = #{catalogId} AND deleted_at = 0"; } + /** Returns SQL that lists all active schemas in a metalake. */ + public String listSchemaPOsByMetalakeId(@Param("metalakeId") Long metalakeId) { + return "SELECT schema_id as schemaId, schema_name as schemaName," + + " metalake_id as metalakeId, catalog_id as catalogId," + + " schema_comment as schemaComment, properties, audit_info as auditInfo," + + " current_version as currentVersion, last_version as lastVersion," + + " deleted_at as deletedAt" + + " FROM " + + TABLE_NAME + + " WHERE metalake_id = #{metalakeId} AND deleted_at = 0"; + } + public String listSchemaPOsByFullQualifiedName( @Param("metalakeName") String metalakeName, @Param("catalogName") String catalogName) { return """ @@ -299,20 +311,26 @@ public String softDeleteSchemaMetasBySchemaIds(@Param("schemaIds") List sc + ""; } - public String softDeleteSchemaMetasByMetalakeId(@Param("metalakeId") Long metalakeId) { + public String softDeleteSchemaMetasByCatalogId(@Param("catalogId") Long catalogId) { return "UPDATE " + TABLE_NAME + " SET deleted_at = (UNIX_TIMESTAMP() * 1000.0)" + " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000" - + " WHERE metalake_id = #{metalakeId} AND deleted_at = 0"; + + " WHERE catalog_id = #{catalogId} AND deleted_at = 0"; } - public String softDeleteSchemaMetasByCatalogId(@Param("catalogId") Long catalogId) { - return "UPDATE " + /** Returns SQL that soft-deletes schemas using identifier-and-version pairs. */ + public String softDeleteSchemaMetasWithVersion(@Param("schemaMetas") List schemaPOs) { + return ""; } public String deleteSchemaMetasByLegacyTimeline( diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/CatalogMetaPostgreSQLProvider.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/CatalogMetaPostgreSQLProvider.java index 0482d9b330b..9ff0849817c 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/CatalogMetaPostgreSQLProvider.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/CatalogMetaPostgreSQLProvider.java @@ -20,6 +20,7 @@ import static org.apache.gravitino.storage.relational.mapper.CatalogMetaMapper.TABLE_NAME; +import java.util.List; import org.apache.gravitino.storage.relational.mapper.provider.base.CatalogMetaBaseSQLProvider; import org.apache.gravitino.storage.relational.po.CatalogPO; import org.apache.ibatis.annotations.Param; @@ -33,12 +34,18 @@ public String softDeleteCatalogMetasByCatalogId(Long catalogId) { + " WHERE catalog_id = #{catalogId} AND deleted_at = 0"; } + /** {@inheritDoc} */ @Override - public String softDeleteCatalogMetasByMetalakeId(Long metalakeId) { - return "UPDATE " + public String softDeleteCatalogMetasWithVersion(List catalogPOs) { + return ""; } @Override diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/MetalakeMetaPostgreSQLProvider.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/MetalakeMetaPostgreSQLProvider.java index 5ce01e67159..20a92d1063c 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/MetalakeMetaPostgreSQLProvider.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/MetalakeMetaPostgreSQLProvider.java @@ -26,11 +26,12 @@ public class MetalakeMetaPostgreSQLProvider extends MetalakeMetaBaseSQLProvider { @Override - public String softDeleteMetalakeMetaByMetalakeId(Long metalakeId) { + public String softDeleteMetalakeMetaByMetalakeId(Long metalakeId, Long currentVersion) { return "UPDATE " + TABLE_NAME + " SET deleted_at = CAST(EXTRACT(EPOCH FROM CURRENT_TIMESTAMP) * 1000 AS BIGINT)" - + " WHERE metalake_id = #{metalakeId} AND deleted_at = 0"; + + " WHERE metalake_id = #{metalakeId}" + + " AND current_version = #{currentVersion} AND deleted_at = 0"; } @Override @@ -75,15 +76,7 @@ public String updateMetalakeMeta( + " current_version = #{newMetalakeMeta.currentVersion}," + " last_version = #{newMetalakeMeta.lastVersion}" + " WHERE metalake_id = #{oldMetalakeMeta.metalakeId}" - + " AND metalake_name = #{oldMetalakeMeta.metalakeName}" - + " AND (metalake_comment = #{oldMetalakeMeta.metalakeComment} " - + " OR (CAST(metalake_comment AS VARCHAR) IS NULL AND " - + " CAST(#{oldMetalakeMeta.metalakeComment} AS VARCHAR) IS NULL))" - + " AND properties = #{oldMetalakeMeta.properties}" - + " AND audit_info = #{oldMetalakeMeta.auditInfo}" - + " AND schema_version = #{oldMetalakeMeta.schemaVersion}" + " AND current_version = #{oldMetalakeMeta.currentVersion}" - + " AND last_version = #{oldMetalakeMeta.lastVersion}" + " AND deleted_at = 0"; } diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/SchemaMetaPostgreSQLProvider.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/SchemaMetaPostgreSQLProvider.java index ba2087aa61a..d805e52c0b4 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/SchemaMetaPostgreSQLProvider.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/SchemaMetaPostgreSQLProvider.java @@ -126,19 +126,25 @@ public String softDeleteSchemaMetasBySchemaIds(List schemaIds) { } @Override - public String softDeleteSchemaMetasByMetalakeId(Long metalakeId) { + public String softDeleteSchemaMetasByCatalogId(Long catalogId) { return "UPDATE " + TABLE_NAME + " SET deleted_at = CAST(EXTRACT(EPOCH FROM CURRENT_TIMESTAMP) * 1000 AS BIGINT)" - + " WHERE metalake_id = #{metalakeId} AND deleted_at = 0"; + + " WHERE catalog_id = #{catalogId} AND deleted_at = 0"; } + /** {@inheritDoc} */ @Override - public String softDeleteSchemaMetasByCatalogId(Long catalogId) { - return "UPDATE " + public String softDeleteSchemaMetasWithVersion(List schemaPOs) { + return ""; } @Override diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/service/MetalakeMetaService.java b/core/src/main/java/org/apache/gravitino/storage/relational/service/MetalakeMetaService.java index ba810a61304..cc42229f73d 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/service/MetalakeMetaService.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/service/MetalakeMetaService.java @@ -25,7 +25,6 @@ import java.io.IOException; import java.util.List; import java.util.Objects; -import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Function; import java.util.stream.Collectors; import org.apache.gravitino.Entity; @@ -34,7 +33,6 @@ import org.apache.gravitino.exceptions.NoSuchEntityException; import org.apache.gravitino.exceptions.NonEmptyEntityException; import org.apache.gravitino.meta.BaseMetalake; -import org.apache.gravitino.meta.CatalogEntity; import org.apache.gravitino.metrics.Monitored; import org.apache.gravitino.storage.relational.mapper.CatalogMetaMapper; import org.apache.gravitino.storage.relational.mapper.FilesetMetaMapper; @@ -64,12 +62,13 @@ import org.apache.gravitino.storage.relational.mapper.UserMetaMapper; import org.apache.gravitino.storage.relational.mapper.UserRoleRelMapper; import org.apache.gravitino.storage.relational.mapper.ViewMetaMapper; +import org.apache.gravitino.storage.relational.po.CatalogPO; import org.apache.gravitino.storage.relational.po.MetalakePO; +import org.apache.gravitino.storage.relational.po.SchemaPO; import org.apache.gravitino.storage.relational.utils.ExceptionUtils; import org.apache.gravitino.storage.relational.utils.POConverters; import org.apache.gravitino.storage.relational.utils.SessionUtils; import org.apache.gravitino.utils.NameIdentifierUtil; -import org.apache.gravitino.utils.NamespaceUtil; /** * The service class for metalake metadata. It provides the basic database operations for metalake. @@ -175,25 +174,27 @@ public BaseMetalake updateMetalake( MetalakePO newMetalakePO = POConverters.updateMetalakePOWithVersion(oldMetalakePO, newMetalakeEntity); - AtomicInteger updateResult = new AtomicInteger(0); try { SessionUtils.doMultipleWithCommit( - () -> - updateResult.set( - SessionUtils.getWithoutCommit( - MetalakeMetaMapper.class, - mapper -> mapper.updateMetalakeMeta(newMetalakePO, oldMetalakePO)))); + () -> { + int updated = + SessionUtils.getWithoutCommit( + MetalakeMetaMapper.class, + mapper -> mapper.updateMetalakeMeta(newMetalakePO, oldMetalakePO)); + if (updated == 0) { + // The row may have a new version, or it may have been deleted or renamed. Re-read it + // to return the correct conflict or missing-entity error. + throw metalakeWriteFailure( + ident, oldMetalakePO.getMetalakeId(), oldMetalakePO.getMetalakeName()); + } + }); } catch (RuntimeException re) { ExceptionUtils.checkSQLException( re, Entity.EntityType.METALAKE, newMetalakeEntity.nameIdentifier().toString()); throw re; } - if (updateResult.get() > 0) { - return newMetalakeEntity; - } else { - throw new IOException("Failed to update the entity: " + ident); - } + return newMetalakeEntity; } @Monitored( @@ -201,22 +202,27 @@ public BaseMetalake updateMetalake( baseMetricName = "deleteMetalake") public boolean deleteMetalake(NameIdentifier ident, boolean cascade) { NameIdentifierUtil.checkMetalake(ident); - Long metalakeId = getMetalakeIdByName(ident.name()); + MetalakePO metalakePO = + SessionUtils.getWithoutCommit( + MetalakeMetaMapper.class, mapper -> mapper.selectMetalakeMetaByName(ident.name())); + if (metalakePO == null) { + throw new NoSuchEntityException( + NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE, + Entity.EntityType.METALAKE.name().toLowerCase(), + ident.toString()); + } + Long metalakeId = metalakePO.getMetalakeId(); + Long currentVersion = metalakePO.getCurrentVersion(); if (metalakeId != null) { if (cascade) { SessionUtils.doMultipleWithCommit( - () -> - SessionUtils.doWithoutCommit( - MetalakeMetaMapper.class, - mapper -> mapper.softDeleteMetalakeMetaByMetalakeId(metalakeId)), - () -> - SessionUtils.doWithoutCommit( - CatalogMetaMapper.class, - mapper -> mapper.softDeleteCatalogMetasByMetalakeId(metalakeId)), - () -> - SessionUtils.doWithoutCommit( - SchemaMetaMapper.class, - mapper -> mapper.softDeleteSchemaMetasByMetalakeId(metalakeId)), + () -> { + // Take the parent lock before the child snapshot, so catalog creation cannot add a + // child after the snapshot. A later failure rolls back this soft delete as well. + deleteMetalakeWithVersion(ident, metalakeId, currentVersion); + deleteCatalogsWithVersions(ident, metalakeId); + deleteSchemasWithVersions(ident, listSchemaPOsForCascade(metalakeId)); + }, () -> SessionUtils.doWithoutCommit( TableMetaMapper.class, @@ -318,18 +324,24 @@ public boolean deleteMetalake(NameIdentifier ident, boolean cascade) { ViewMetaMapper.class, mapper -> mapper.softDeleteViewMetasByMetalakeId(metalakeId))); } else { - List catalogEntities = - CatalogMetaService.getInstance() - .listCatalogsByNamespace(NamespaceUtil.ofCatalog(ident.name())); - if (!catalogEntities.isEmpty()) { - throw new NonEmptyEntityException( - "Entity %s has sub-entities, you should remove sub-entities first", ident); - } SessionUtils.doMultipleWithCommit( - () -> - SessionUtils.doWithoutCommit( - MetalakeMetaMapper.class, - mapper -> mapper.softDeleteMetalakeMetaByMetalakeId(metalakeId)), + () -> { + // Delete the metalake before checking its children. The UPDATE takes an exclusive + // lock on the metalake row, and catalog creation locks the same row before inserting. + // If creation gets its lock first, this delete waits and the check below sees the new + // catalog. If this delete gets the lock first, creation waits until this transaction + // commits or rolls back. Checking first would allow a catalog to be inserted between + // the check and this delete. A non-empty result rolls back the soft delete. + deleteMetalakeWithVersion(ident, metalakeId, currentVersion); + List catalogPOs = + SessionUtils.getWithoutCommit( + CatalogMetaMapper.class, + mapper -> mapper.listCatalogPOsByMetalakeId(metalakeId)); + if (!catalogPOs.isEmpty()) { + throw new NonEmptyEntityException( + "Entity %s has sub-entities, you should remove sub-entities first", ident); + } + }, () -> SessionUtils.doWithoutCommit( UserRoleRelMapper.class, @@ -383,6 +395,81 @@ public boolean deleteMetalake(NameIdentifier ident, boolean cascade) { return true; } + void deleteMetalakeWithVersion(NameIdentifier identifier, Long metalakeId, Long currentVersion) { + int deleted = + SessionUtils.getWithoutCommit( + MetalakeMetaMapper.class, + mapper -> mapper.softDeleteMetalakeMetaByMetalakeId(metalakeId, currentVersion)); + if (deleted == 0) { + throw metalakeWriteFailure(identifier, metalakeId, identifier.name()); + } + } + + private RuntimeException metalakeWriteFailure( + NameIdentifier identifier, Long metalakeId, String observedName) { + // Use a locking read to see the latest committed row. Under MySQL REPEATABLE READ, a plain + // SELECT can return an old snapshot that still contains a row another writer already deleted + // or renamed. We would then report a version conflict instead of a missing metalake. The CAS + // UPDATE above already waits for the same row lock, so the other writer has finished before + // this read runs. + MetalakePO currentMetalakePO = + SessionUtils.getWithoutCommit( + MetalakeMetaMapper.class, mapper -> mapper.selectMetalakeMetaByIdForUpdate(metalakeId)); + if (currentMetalakePO == null + || !Objects.equals(currentMetalakePO.getMetalakeName(), observedName)) { + return new NoSuchEntityException( + NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE, + Entity.EntityType.METALAKE.name().toLowerCase(), + identifier.name()); + } + return ExceptionUtils.concurrentModification(Entity.EntityType.METALAKE, identifier); + } + + private void deleteCatalogsWithVersions(NameIdentifier metalakeIdentifier, Long metalakeId) { + // Lock all catalog rows before taking the schema snapshot. Schema creation and deletion lock + // their parent catalog, so they cannot add or remove a schema after this point. A schema alter + // can still run, but the version check below detects it. Taking parent locks before child rows + // also keeps the same lock order for every metalake cascade. + List catalogPOs = + SessionUtils.getWithoutCommit( + CatalogMetaMapper.class, + mapper -> mapper.listCatalogPOsByMetalakeIdForUpdate(metalakeId)); + if (catalogPOs.isEmpty()) { + return; + } + int deleted = + SessionUtils.getWithoutCommit( + CatalogMetaMapper.class, + mapper -> mapper.softDeleteCatalogMetasWithVersion(catalogPOs)); + // Never commit a partial cascade. A smaller count means that a catalog no longer matches the + // ID and version read above, so the outer transaction must roll back all deletes. + if (deleted != catalogPOs.size()) { + throw ExceptionUtils.concurrentChildModification( + Entity.EntityType.CATALOG, Entity.EntityType.METALAKE, metalakeIdentifier); + } + } + + List listSchemaPOsForCascade(Long metalakeId) { + return SessionUtils.getWithoutCommit( + SchemaMetaMapper.class, mapper -> mapper.listSchemaPOsByMetalakeId(metalakeId)); + } + + private void deleteSchemasWithVersions( + NameIdentifier metalakeIdentifier, List schemaPOs) { + if (schemaPOs.isEmpty()) { + return; + } + int deleted = + SessionUtils.getWithoutCommit( + SchemaMetaMapper.class, mapper -> mapper.softDeleteSchemaMetasWithVersion(schemaPOs)); + // The version check protects this snapshot from a schema alter that does not use the parent + // catalog lock. Roll back the whole cascade instead of silently losing that schema change. + if (deleted != schemaPOs.size()) { + throw ExceptionUtils.concurrentChildModification( + Entity.EntityType.SCHEMA, Entity.EntityType.METALAKE, metalakeIdentifier); + } + } + @Monitored( metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME, baseMetricName = "deleteMetalakeMetasByLegacyTimeline") diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/utils/ExceptionUtils.java b/core/src/main/java/org/apache/gravitino/storage/relational/utils/ExceptionUtils.java index eb08cfd2e89..7da30085003 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/utils/ExceptionUtils.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/utils/ExceptionUtils.java @@ -20,7 +20,10 @@ import java.io.IOException; import java.sql.SQLException; +import java.util.Locale; import org.apache.gravitino.Entity; +import org.apache.gravitino.NameIdentifier; +import org.apache.gravitino.exceptions.OptimisticLockException; import org.apache.gravitino.storage.relational.converters.SQLExceptionConverterFactory; public class ExceptionUtils { @@ -33,4 +36,37 @@ public static void checkSQLException( .toGravitinoException((SQLException) re.getCause(), type, entityName); } } + + /** + * Creates an {@link OptimisticLockException} for an entity that was modified concurrently by + * another writer, which makes the version-guarded write fail. + * + * @param type The type of the entity that was modified concurrently. + * @param identifier The identifier of the entity that was modified concurrently. + * @return The {@link OptimisticLockException} to throw. + */ + public static OptimisticLockException concurrentModification( + Entity.EntityType type, NameIdentifier identifier) { + return new OptimisticLockException( + "The %s %s was modified concurrently; retry the operation", + type.name().toLowerCase(Locale.ROOT), identifier); + } + + /** + * Creates an {@link OptimisticLockException} for a child entity that was modified concurrently + * while its parent was being deleted in cascade mode. + * + * @param childType The type of the child entity that was modified concurrently. + * @param parentType The type of the parent entity being operated on. + * @param parentIdentifier The identifier of the parent entity being operated on. + * @return The {@link OptimisticLockException} to throw. + */ + public static OptimisticLockException concurrentChildModification( + Entity.EntityType childType, Entity.EntityType parentType, NameIdentifier parentIdentifier) { + return new OptimisticLockException( + "A %s under %s %s was modified concurrently; retry the operation", + childType.name().toLowerCase(Locale.ROOT), + parentType.name().toLowerCase(Locale.ROOT), + parentIdentifier); + } } diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/utils/POConverters.java b/core/src/main/java/org/apache/gravitino/storage/relational/utils/POConverters.java index d019800d1ab..2a05888fd7d 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/utils/POConverters.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/utils/POConverters.java @@ -137,9 +137,9 @@ public static MetalakePO initializeMetalakePOWithVersion(BaseMetalake baseMetala */ public static MetalakePO updateMetalakePOWithVersion( MetalakePO oldMetalakePO, BaseMetalake newMetalake) { - Long lastVersion = oldMetalakePO.getLastVersion(); - // Will set the version to the last version + 1 when having some fields need be multiple version - Long nextVersion = lastVersion; + // Every metadata update advances the OCC token. Both version columns stay aligned because + // metalakes do not retain independently addressable historical versions. + Long nextVersion = oldMetalakePO.getCurrentVersion() + 1; try { return MetalakePO.builder() .withMetalakeId(newMetalake.id()) diff --git a/core/src/test/java/org/apache/gravitino/metalake/TestMetalakeManager.java b/core/src/test/java/org/apache/gravitino/metalake/TestMetalakeManager.java index c5be0774ea8..55084847f33 100644 --- a/core/src/test/java/org/apache/gravitino/metalake/TestMetalakeManager.java +++ b/core/src/test/java/org/apache/gravitino/metalake/TestMetalakeManager.java @@ -31,6 +31,7 @@ import java.util.Set; import org.apache.commons.lang3.reflect.FieldUtils; import org.apache.gravitino.Config; +import org.apache.gravitino.Entity.EntityType; import org.apache.gravitino.EntityStore; import org.apache.gravitino.GravitinoEnv; import org.apache.gravitino.MetalakeChange; @@ -39,11 +40,13 @@ import org.apache.gravitino.UserPrincipal; import org.apache.gravitino.auth.AuthConstants; import org.apache.gravitino.exceptions.MetalakeAlreadyExistsException; +import org.apache.gravitino.exceptions.NoSuchEntityException; import org.apache.gravitino.exceptions.NoSuchMetalakeException; import org.apache.gravitino.lock.LockManager; import org.apache.gravitino.meta.BaseMetalake; import org.apache.gravitino.storage.RandomIdGenerator; import org.apache.gravitino.storage.memory.TestMemoryEntityStore; +import org.apache.gravitino.storage.memory.TestMemoryEntityStore.InMemoryEntityStore; import org.apache.gravitino.utils.PrincipalUtils; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.Assertions; @@ -210,6 +213,23 @@ public void testDropMetalake() { Assertions.assertFalse(dropped1, "metalake should be non-existent"); } + @Test + public void testDropMetalakeReturnsFalseWhenConcurrentDeleteWins() throws IOException { + InMemoryEntityStore store = Mockito.spy(new InMemoryEntityStore()); + store.initialize(config); + MetalakeManager manager = new MetalakeManager(store, new RandomIdGenerator()); + NameIdentifier ident = NameIdentifier.of("concurrently_deleted_metalake"); + manager.createMetalake(ident, "comment", ImmutableMap.of()); + Mockito.doThrow( + new NoSuchEntityException( + NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE, "metalake", ident.toString())) + .when(store) + .delete(ident, EntityType.METALAKE, true); + + Assertions.assertFalse(manager.dropMetalake(ident, true)); + store.close(); + } + @Test public void testListInUseMetalakes() { // Create some metalakes with different in-use status diff --git a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestMetalakeMetaService.java b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestMetalakeMetaService.java index b1e6389a208..7948c75a9cf 100644 --- a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestMetalakeMetaService.java +++ b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestMetalakeMetaService.java @@ -27,10 +27,17 @@ import java.util.List; import org.apache.gravitino.Entity; import org.apache.gravitino.EntityAlreadyExistsException; +import org.apache.gravitino.exceptions.NoSuchEntityException; +import org.apache.gravitino.exceptions.NonEmptyEntityException; +import org.apache.gravitino.exceptions.OptimisticLockException; import org.apache.gravitino.meta.BaseMetalake; import org.apache.gravitino.meta.SchemaVersion; import org.apache.gravitino.storage.RandomIdGenerator; import org.apache.gravitino.storage.relational.TestJDBCBackend; +import org.apache.gravitino.storage.relational.mapper.MetalakeMetaMapper; +import org.apache.gravitino.storage.relational.po.MetalakePO; +import org.apache.gravitino.storage.relational.utils.POConverters; +import org.apache.gravitino.storage.relational.utils.SessionUtils; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.TestTemplate; @@ -92,6 +99,172 @@ void testUpdateMetalakeWithNullableComment() throws IOException { backend.delete(metalake.nameIdentifier(), Entity.EntityType.METALAKE, false); } + @TestTemplate + public void testAlterAndDeleteUseCurrentVersion() throws IOException { + BaseMetalake metalake = createAndInsertMakeLake(METALAKE_NAME); + MetalakePO oldPO = + SessionUtils.getWithoutCommit( + MetalakeMetaMapper.class, mapper -> mapper.selectMetalakeMetaByName(metalake.name())); + BaseMetalake updatedMetalake = + BaseMetalake.builder() + .withId(metalake.id()) + .withName(metalake.name()) + .withAuditInfo(metalake.auditInfo()) + .withComment("updated") + .withProperties(metalake.properties()) + .withVersion(metalake.getVersion()) + .build(); + MetalakePO newPO = POConverters.updateMetalakePOWithVersion(oldPO, updatedMetalake); + + int updated = + SessionUtils.doWithCommitAndFetchResult( + MetalakeMetaMapper.class, mapper -> mapper.updateMetalakeMeta(newPO, oldPO)); + int staleUpdate = + SessionUtils.doWithCommitAndFetchResult( + MetalakeMetaMapper.class, mapper -> mapper.updateMetalakeMeta(newPO, oldPO)); + int staleDelete = + SessionUtils.doWithCommitAndFetchResult( + MetalakeMetaMapper.class, + mapper -> + mapper.softDeleteMetalakeMetaByMetalakeId( + metalake.id(), oldPO.getCurrentVersion())); + Assertions.assertEquals(1, updated); + Assertions.assertEquals(0, staleUpdate); + Assertions.assertEquals(0, staleDelete); + assertTrue(backend.exists(metalake.nameIdentifier(), Entity.EntityType.METALAKE)); + int deleted = + SessionUtils.doWithCommitAndFetchResult( + MetalakeMetaMapper.class, + mapper -> + mapper.softDeleteMetalakeMetaByMetalakeId( + metalake.id(), newPO.getCurrentVersion())); + Assertions.assertEquals(1, deleted); + } + + @TestTemplate + public void testAlterReportsOptimisticLockConflict() throws IOException { + BaseMetalake metalake = createAndInsertMakeLake(METALAKE_NAME); + + assertThrows( + OptimisticLockException.class, + () -> + MetalakeMetaService.getInstance() + .updateMetalake( + metalake.nameIdentifier(), + entity -> { + BaseMetalake current = (BaseMetalake) entity; + MetalakePO currentPO = + SessionUtils.getWithoutCommit( + MetalakeMetaMapper.class, + mapper -> mapper.selectMetalakeMetaByName(current.name())); + BaseMetalake competingUpdate = + BaseMetalake.builder() + .withId(current.id()) + .withName(current.name()) + .withAuditInfo(current.auditInfo()) + .withComment("competing update") + .withProperties(current.properties()) + .withVersion(current.getVersion()) + .build(); + MetalakePO competingPO = + POConverters.updateMetalakePOWithVersion(currentPO, competingUpdate); + SessionUtils.doWithCommitAndFetchResult( + MetalakeMetaMapper.class, + mapper -> mapper.updateMetalakeMeta(competingPO, currentPO)); + return BaseMetalake.builder() + .withId(current.id()) + .withName(current.name()) + .withAuditInfo(current.auditInfo()) + .withComment("requested update") + .withProperties(current.properties()) + .withVersion(current.getVersion()) + .build(); + })); + } + + @TestTemplate + public void testAlterReportsNoSuchWhenMetalakeIsDeletedConcurrently() throws IOException { + BaseMetalake metalake = createAndInsertMakeLake(METALAKE_NAME); + + assertThrows( + NoSuchEntityException.class, + () -> + MetalakeMetaService.getInstance() + .updateMetalake( + metalake.nameIdentifier(), + entity -> { + BaseMetalake current = (BaseMetalake) entity; + MetalakePO currentPO = + SessionUtils.getWithoutCommit( + MetalakeMetaMapper.class, + mapper -> mapper.selectMetalakeMetaById(current.id())); + SessionUtils.doWithCommitAndFetchResult( + MetalakeMetaMapper.class, + mapper -> + mapper.softDeleteMetalakeMetaByMetalakeId( + current.id(), currentPO.getCurrentVersion())); + return BaseMetalake.builder() + .withId(current.id()) + .withName(current.name()) + .withAuditInfo(current.auditInfo()) + .withComment("requested update") + .withProperties(current.properties()) + .withVersion(current.getVersion()) + .build(); + })); + } + + @TestTemplate + public void testDeleteReportsOptimisticLockConflict() throws IOException { + BaseMetalake metalake = createAndInsertMakeLake(METALAKE_NAME); + MetalakePO stalePO = + SessionUtils.getWithoutCommit( + MetalakeMetaMapper.class, mapper -> mapper.selectMetalakeMetaByName(metalake.name())); + BaseMetalake competingUpdate = + BaseMetalake.builder() + .withId(metalake.id()) + .withName(metalake.name()) + .withAuditInfo(metalake.auditInfo()) + .withComment("competing update") + .withProperties(metalake.properties()) + .withVersion(metalake.getVersion()) + .build(); + MetalakePO competingPO = POConverters.updateMetalakePOWithVersion(stalePO, competingUpdate); + SessionUtils.doWithCommitAndFetchResult( + MetalakeMetaMapper.class, mapper -> mapper.updateMetalakeMeta(competingPO, stalePO)); + + assertThrows( + OptimisticLockException.class, + () -> + SessionUtils.doMultipleWithCommit( + () -> + MetalakeMetaService.getInstance() + .deleteMetalakeWithVersion( + metalake.nameIdentifier(), + metalake.id(), + stalePO.getCurrentVersion()))); + assertTrue(backend.exists(metalake.nameIdentifier(), Entity.EntityType.METALAKE)); + } + + @TestTemplate + public void testNonCascadeDeleteRollsBackMetalakeFence() throws IOException { + BaseMetalake metalake = createAndInsertMakeLake(METALAKE_NAME); + createAndInsertCatalog(METALAKE_NAME, "catalog"); + MetalakePO beforeDelete = + SessionUtils.getWithoutCommit( + MetalakeMetaMapper.class, mapper -> mapper.selectMetalakeMetaByName(metalake.name())); + + assertThrows( + NonEmptyEntityException.class, + () -> MetalakeMetaService.getInstance().deleteMetalake(metalake.nameIdentifier(), false)); + + MetalakePO afterDelete = + SessionUtils.getWithoutCommit( + MetalakeMetaMapper.class, mapper -> mapper.selectMetalakeMetaByName(metalake.name())); + Assertions.assertEquals(beforeDelete.getCurrentVersion(), afterDelete.getCurrentVersion()); + assertTrue(backend.exists(metalake.nameIdentifier(), Entity.EntityType.METALAKE)); + } + @TestTemplate public void testMetaLifeCycleFromCreationToDeletion() throws IOException { // meta data creation diff --git a/core/src/test/java/org/apache/gravitino/storage/relational/utils/TestExceptionUtils.java b/core/src/test/java/org/apache/gravitino/storage/relational/utils/TestExceptionUtils.java new file mode 100644 index 00000000000..0d489be6882 --- /dev/null +++ b/core/src/test/java/org/apache/gravitino/storage/relational/utils/TestExceptionUtils.java @@ -0,0 +1,50 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.gravitino.storage.relational.utils; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import org.apache.gravitino.Entity; +import org.apache.gravitino.NameIdentifier; +import org.apache.gravitino.exceptions.OptimisticLockException; +import org.junit.jupiter.api.Test; + +public class TestExceptionUtils { + + @Test + public void testConcurrentModificationMessage() { + OptimisticLockException e = + ExceptionUtils.concurrentModification( + Entity.EntityType.CATALOG, NameIdentifier.of("m1", "c1")); + + assertEquals( + "The catalog m1.c1 was modified concurrently; retry the operation", e.getMessage()); + } + + @Test + public void testConcurrentChildModificationMessage() { + OptimisticLockException e = + ExceptionUtils.concurrentChildModification( + Entity.EntityType.SCHEMA, Entity.EntityType.CATALOG, NameIdentifier.of("m1", "c1")); + + assertEquals( + "A schema under catalog m1.c1 was modified concurrently; retry the operation", + e.getMessage()); + } +} diff --git a/core/src/test/java/org/apache/gravitino/storage/relational/utils/TestPOConverters.java b/core/src/test/java/org/apache/gravitino/storage/relational/utils/TestPOConverters.java index 9ce171d8782..22a53895065 100644 --- a/core/src/test/java/org/apache/gravitino/storage/relational/utils/TestPOConverters.java +++ b/core/src/test/java/org/apache/gravitino/storage/relational/utils/TestPOConverters.java @@ -665,6 +665,8 @@ public void testUpdateMetalakePOVersion() { assertEquals(1, initPO.getCurrentVersion()); assertEquals(1, initPO.getLastVersion()); assertEquals(0, initPO.getDeletedAt()); + assertEquals(2, updatePO.getCurrentVersion()); + assertEquals(2, updatePO.getLastVersion()); assertEquals("this is test2", updatePO.getMetalakeComment()); }