diff --git a/core/src/main/java/org/apache/gravitino/catalog/ManagedSchemaOperations.java b/core/src/main/java/org/apache/gravitino/catalog/ManagedSchemaOperations.java index 164d4b53460..aca4eec619a 100644 --- a/core/src/main/java/org/apache/gravitino/catalog/ManagedSchemaOperations.java +++ b/core/src/main/java/org/apache/gravitino/catalog/ManagedSchemaOperations.java @@ -25,6 +25,7 @@ import java.util.List; import java.util.Map; import org.apache.gravitino.Entity; +import org.apache.gravitino.EntityAlreadyExistsException; import org.apache.gravitino.EntityStore; import org.apache.gravitino.NameIdentifier; import org.apache.gravitino.Namespace; @@ -116,9 +117,11 @@ public Schema createSchema(NameIdentifier ident, String comment, Map catalogPOs); @UpdateProvider( 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..86467f6cef2 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 @@ -109,8 +109,21 @@ public static String updateCatalogMeta( return getProvider().updateCatalogMeta(newCatalogPO, oldCatalogPO); } - public static String softDeleteCatalogMetasByCatalogId(@Param("catalogId") Long catalogId) { - return getProvider().softDeleteCatalogMetasByCatalogId(catalogId); + /** Returns SQL that advances a catalog OCC version conditionally. */ + public static String fenceCatalogMeta( + @Param("catalogId") Long catalogId, @Param("currentVersion") Long currentVersion) { + return getProvider().fenceCatalogMeta(catalogId, currentVersion); + } + + public static String softDeleteCatalogMetasByCatalogId( + @Param("catalogId") Long catalogId, @Param("currentVersion") Long currentVersion) { + return getProvider().softDeleteCatalogMetasByCatalogId(catalogId, currentVersion); + } + + /** 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 softDeleteCatalogMetasByMetalakeId(@Param("metalakeId") Long metalakeId) { diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/GroupMetaMapper.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/GroupMetaMapper.java index b74116fb24f..2a045ea5745 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/GroupMetaMapper.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/GroupMetaMapper.java @@ -76,7 +76,8 @@ List listExtendedGroupPOsByMetalakeIdAndNames( void insertGroupMetaOnDuplicateKeyUpdate(@Param("groupMeta") GroupPO groupPO); @UpdateProvider(type = GroupMetaSQLProviderFactory.class, method = "softDeleteGroupMetaByGroupId") - void softDeleteGroupMetaByGroupId(@Param("groupId") Long groupId); + Integer softDeleteGroupMetaByGroupId( + @Param("groupId") Long groupId, @Param("currentVersion") Long currentVersion); @UpdateProvider( type = GroupMetaSQLProviderFactory.class, diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/GroupMetaSQLProviderFactory.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/GroupMetaSQLProviderFactory.java index 0981d4b401f..3123bb8a6a9 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/GroupMetaSQLProviderFactory.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/GroupMetaSQLProviderFactory.java @@ -72,8 +72,9 @@ public static String insertGroupMetaOnDuplicateKeyUpdate(@Param("groupMeta") Gro return getProvider().insertGroupMetaOnDuplicateKeyUpdate(groupPO); } - public static String softDeleteGroupMetaByGroupId(@Param("groupId") Long groupId) { - return getProvider().softDeleteGroupMetaByGroupId(groupId); + public static String softDeleteGroupMetaByGroupId( + @Param("groupId") Long groupId, @Param("currentVersion") Long currentVersion) { + return getProvider().softDeleteGroupMetaByGroupId(groupId, currentVersion); } public static String softDeleteGroupMetasByMetalakeId(@Param("metalakeId") Long metalakeId) { 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..4b40bc751ea 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 @@ -70,10 +70,20 @@ Integer updateMetalakeMeta( @Param("newMetalakeMeta") MetalakePO newMetalakePO, @Param("oldMetalakeMeta") MetalakePO oldMetalakePO); + /** + * Advances the metalake version when the expected OCC version still matches. + * + * @return the number of updated rows + */ + @UpdateProvider(type = MetalakeMetaSQLProviderFactory.class, method = "fenceMetalakeMeta") + Integer fenceMetalakeMeta( + @Param("metalakeId") Long metalakeId, @Param("currentVersion") Long currentVersion); + @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..79a91f41f7e 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 @@ -88,8 +88,15 @@ public static String updateMetalakeMeta( return getProvider().updateMetalakeMeta(newMetalakePO, oldMetalakePO); } - public static String softDeleteMetalakeMetaByMetalakeId(@Param("metalakeId") Long metalakeId) { - return getProvider().softDeleteMetalakeMetaByMetalakeId(metalakeId); + /** Returns SQL that advances a metalake OCC version conditionally. */ + public static String fenceMetalakeMeta( + @Param("metalakeId") Long metalakeId, @Param("currentVersion") Long currentVersion) { + return getProvider().fenceMetalakeMeta(metalakeId, currentVersion); + } + + 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..532a0b7d247 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 @@ -102,11 +102,36 @@ SchemaPO selectSchemaByFullQualifiedName( Integer updateSchemaMeta( @Param("newSchemaMeta") SchemaPO newSchemaPO, @Param("oldSchemaMeta") SchemaPO oldSchemaPO); + /** + * Advances the schema version when the expected OCC version still matches. + * + * @return the number of updated rows + */ + @UpdateProvider(type = SchemaMetaSQLProviderFactory.class, method = "fenceSchemaMeta") + Integer fenceSchemaMeta( + @Param("schemaId") Long schemaId, @Param("currentVersion") Long currentVersion); + @UpdateProvider( type = SchemaMetaSQLProviderFactory.class, method = "softDeleteSchemaMetasBySchemaIds") Integer softDeleteSchemaMetasBySchemaIds(@Param("schemaIds") List schemaIds); + @UpdateProvider( + type = SchemaMetaSQLProviderFactory.class, + method = "softDeleteSchemaMetaBySchemaIdAndVersion") + Integer softDeleteSchemaMetaBySchemaIdAndVersion( + @Param("schemaId") Long schemaId, @Param("currentVersion") Long currentVersion); + + /** + * Soft-deletes schemas whose identifiers and OCC versions still match. + * + * @return the number of deleted rows + */ + @UpdateProvider( + type = SchemaMetaSQLProviderFactory.class, + method = "softDeleteSchemaMetasWithVersion") + Integer softDeleteSchemaMetasWithVersion(@Param("schemaMetas") List schemaPOs); + @UpdateProvider( type = SchemaMetaSQLProviderFactory.class, method = "softDeleteSchemaMetasByMetalakeId") 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..ba368c219fa 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 @@ -116,10 +116,27 @@ public static String updateSchemaMeta( return getProvider().updateSchemaMeta(newSchemaPO, oldSchemaPO); } + /** Returns SQL that advances a schema OCC version conditionally. */ + public static String fenceSchemaMeta( + @Param("schemaId") Long schemaId, @Param("currentVersion") Long currentVersion) { + return getProvider().fenceSchemaMeta(schemaId, currentVersion); + } + public static String softDeleteSchemaMetasBySchemaIds(@Param("schemaIds") List schemaIds) { return getProvider().softDeleteSchemaMetasBySchemaIds(schemaIds); } + public static String softDeleteSchemaMetaBySchemaIdAndVersion( + @Param("schemaId") Long schemaId, @Param("currentVersion") Long currentVersion) { + return getProvider().softDeleteSchemaMetaBySchemaIdAndVersion(schemaId, currentVersion); + } + + /** Returns SQL that soft-deletes schemas using identifier-and-version pairs. */ + public static String softDeleteSchemaMetasWithVersion( + @Param("schemaMetas") List schemaPOs) { + return getProvider().softDeleteSchemaMetasWithVersion(schemaPOs); + } + public static String softDeleteSchemaMetasByMetalakeId(@Param("metalakeId") Long metalakeId) { return getProvider().softDeleteSchemaMetasByMetalakeId(metalakeId); } diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/UserMetaMapper.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/UserMetaMapper.java index 45bcb63c5db..771802d221c 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/UserMetaMapper.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/UserMetaMapper.java @@ -71,7 +71,8 @@ UserPO selectUserMetaByMetalakeIdAndName( void insertUserMetaOnDuplicateKeyUpdate(@Param("userMeta") UserPO userPO); @UpdateProvider(type = UserMetaSQLProviderFactory.class, method = "softDeleteUserMetaByUserId") - void softDeleteUserMetaByUserId(@Param("userId") Long userId); + Integer softDeleteUserMetaByUserId( + @Param("userId") Long userId, @Param("currentVersion") Long currentVersion); @UpdateProvider( type = UserMetaSQLProviderFactory.class, diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/UserMetaSQLProviderFactory.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/UserMetaSQLProviderFactory.java index 9d668dd2f34..db300f58d6a 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/UserMetaSQLProviderFactory.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/UserMetaSQLProviderFactory.java @@ -69,8 +69,9 @@ public static String insertUserMetaOnDuplicateKeyUpdate(@Param("userMeta") UserP return getProvider().insertUserMetaOnDuplicateKeyUpdate(userPO); } - public static String softDeleteUserMetaByUserId(@Param("userId") Long userId) { - return getProvider().softDeleteUserMetaByUserId(userId); + public static String softDeleteUserMetaByUserId( + @Param("userId") Long userId, @Param("currentVersion") Long currentVersion) { + return getProvider().softDeleteUserMetaByUserId(userId, currentVersion); } public static String softDeleteUserMetasByMetalakeId(@Param("metalakeId") Long metalakeId) { 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..bdaee81ac3b 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 @@ -206,25 +206,43 @@ public String updateCatalogMeta( + " last_version = #{newCatalogMeta.lastVersion}," + " deleted_at = #{newCatalogMeta.deletedAt}" + " WHERE catalog_id = #{oldCatalogMeta.catalogId}" - + " AND catalog_name = #{oldCatalogMeta.catalogName}" - + " AND metalake_id = #{oldCatalogMeta.metalakeId}" - + " AND type = #{oldCatalogMeta.type}" - + " AND provider = #{oldCatalogMeta.provider}" - + " AND (catalog_comment = #{oldCatalogMeta.catalogComment} " - + " OR (catalog_comment IS NULL and #{oldCatalogMeta.catalogComment} IS NULL))" - + " AND properties = #{oldCatalogMeta.properties}" - + " AND audit_info = #{oldCatalogMeta.auditInfo}" + " AND current_version = #{oldCatalogMeta.currentVersion}" - + " AND last_version = #{oldCatalogMeta.lastVersion}" + " AND deleted_at = 0"; } - public String softDeleteCatalogMetasByCatalogId(@Param("catalogId") Long catalogId) { + /** Returns SQL that advances a catalog OCC version conditionally. */ + public String fenceCatalogMeta( + @Param("catalogId") Long catalogId, @Param("currentVersion") Long currentVersion) { + return "UPDATE " + + TABLE_NAME + + " SET last_version = current_version + 1, current_version = current_version + 1" + + " WHERE catalog_id = #{catalogId}" + + " AND current_version = #{currentVersion} AND deleted_at = 0"; + } + + public String softDeleteCatalogMetasByCatalogId( + @Param("catalogId") Long catalogId, @Param("currentVersion") Long currentVersion) { return "UPDATE " + TABLE_NAME + " SET deleted_at = (UNIX_TIMESTAMP() * 1000.0)" + " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000" - + " WHERE catalog_id = #{catalogId} AND deleted_at = 0"; + + " WHERE catalog_id = #{catalogId}" + + " AND current_version = #{currentVersion} AND deleted_at = 0"; + } + + /** Returns SQL that soft-deletes catalogs using identifier-and-version pairs. */ + public String softDeleteCatalogMetasWithVersion( + @Param("catalogMetas") List catalogPOs) { + return ""; } public String softDeleteCatalogMetasByMetalakeId(@Param("metalakeId") Long metalakeId) { diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/GroupMetaBaseSQLProvider.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/GroupMetaBaseSQLProvider.java index fa9085a4293..77aefe0d7fb 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/GroupMetaBaseSQLProvider.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/GroupMetaBaseSQLProvider.java @@ -201,12 +201,14 @@ public String insertGroupMetaOnDuplicateKeyUpdate(@Param("groupMeta") GroupPO gr + " deleted_at = #{groupMeta.deletedAt}"; } - public String softDeleteGroupMetaByGroupId(@Param("groupId") Long groupId) { + public String softDeleteGroupMetaByGroupId( + @Param("groupId") Long groupId, @Param("currentVersion") Long currentVersion) { return "UPDATE " + GROUP_TABLE_NAME + " SET deleted_at = (UNIX_TIMESTAMP() * 1000.0)" + " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000" - + " WHERE group_id = #{groupId} AND deleted_at = 0"; + + " WHERE group_id = #{groupId}" + + " AND current_version = #{currentVersion} AND deleted_at = 0"; } public String softDeleteGroupMetasByMetalakeId(@Param("metalakeId") Long metalakeId) { @@ -229,11 +231,7 @@ public String updateGroupMeta( + " last_version = #{newGroupMeta.lastVersion}," + " deleted_at = #{newGroupMeta.deletedAt}" + " WHERE group_id = #{oldGroupMeta.groupId}" - + " AND group_name = #{oldGroupMeta.groupName}" - + " AND metalake_id = #{oldGroupMeta.metalakeId}" - + " AND audit_info = #{oldGroupMeta.auditInfo}" + " AND current_version = #{oldGroupMeta.currentVersion}" - + " AND last_version = #{oldGroupMeta.lastVersion}" + " AND deleted_at = 0"; } 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..5899b3d42b9 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 @@ -143,23 +143,28 @@ 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) { + /** Returns SQL that advances a metalake OCC version conditionally. */ + public String fenceMetalakeMeta( + @Param("metalakeId") Long metalakeId, @Param("currentVersion") Long currentVersion) { + return "UPDATE " + + TABLE_NAME + + " SET last_version = current_version + 1, current_version = current_version + 1" + + " WHERE metalake_id = #{metalakeId}" + + " AND current_version = #{currentVersion} AND deleted_at = 0"; + } + + 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..df7fd809586 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 @@ -273,18 +273,20 @@ public String updateSchemaMeta( + " last_version = #{newSchemaMeta.lastVersion}," + " deleted_at = #{newSchemaMeta.deletedAt}" + " WHERE schema_id = #{oldSchemaMeta.schemaId}" - + " AND schema_name = #{oldSchemaMeta.schemaName}" - + " AND metalake_id = #{oldSchemaMeta.metalakeId}" - + " AND catalog_id = #{oldSchemaMeta.catalogId}" - + " AND (schema_comment = #{oldSchemaMeta.schemaComment}" - + " OR (schema_comment IS NULL and #{oldSchemaMeta.schemaComment} IS NULL))" - + " AND properties = #{oldSchemaMeta.properties}" - + " AND audit_info = #{oldSchemaMeta.auditInfo}" + " AND current_version = #{oldSchemaMeta.currentVersion}" - + " AND last_version = #{oldSchemaMeta.lastVersion}" + " AND deleted_at = 0"; } + /** Returns SQL that advances a schema OCC version conditionally. */ + public String fenceSchemaMeta( + @Param("schemaId") Long schemaId, @Param("currentVersion") Long currentVersion) { + return "UPDATE " + + TABLE_NAME + + " SET last_version = current_version + 1, current_version = current_version + 1" + + " WHERE schema_id = #{schemaId}" + + " AND current_version = #{currentVersion} AND deleted_at = 0"; + } + public String softDeleteSchemaMetasBySchemaIds(@Param("schemaIds") List schemaIds) { return ""; } + public String softDeleteSchemaMetaBySchemaIdAndVersion( + @Param("schemaId") Long schemaId, @Param("currentVersion") Long currentVersion) { + return "UPDATE " + + TABLE_NAME + + " SET deleted_at = (UNIX_TIMESTAMP() * 1000.0)" + + " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000" + + " WHERE schema_id = #{schemaId}" + + " AND current_version = #{currentVersion} AND deleted_at = 0"; + } + + /** Returns SQL that soft-deletes schemas using identifier-and-version pairs. */ + public String softDeleteSchemaMetasWithVersion(@Param("schemaMetas") List schemaPOs) { + return ""; + } + public String softDeleteSchemaMetasByMetalakeId(@Param("metalakeId") Long metalakeId) { return "UPDATE " + TABLE_NAME diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/UserMetaBaseSQLProvider.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/UserMetaBaseSQLProvider.java index c0647d38557..e6587df3ace 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/UserMetaBaseSQLProvider.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/UserMetaBaseSQLProvider.java @@ -99,11 +99,8 @@ public String updateUserMetaByExternalId( + " current_version = #{newUserMeta.currentVersion}," + " last_version = #{newUserMeta.lastVersion}," + " deleted_at = #{newUserMeta.deletedAt}" - + " WHERE external_id = #{oldUserMeta.externalId}" - + " AND metalake_id = #{oldUserMeta.metalakeId}" - + " AND audit_info = #{oldUserMeta.auditInfo}" + + " WHERE user_id = #{oldUserMeta.userId}" + " AND current_version = #{oldUserMeta.currentVersion}" - + " AND last_version = #{oldUserMeta.lastVersion}" + " AND deleted_at = 0"; } @@ -152,12 +149,14 @@ public String insertUserMetaOnDuplicateKeyUpdate(@Param("userMeta") UserPO userP + " deleted_at = #{userMeta.deletedAt}"; } - public String softDeleteUserMetaByUserId(@Param("userId") Long userId) { + public String softDeleteUserMetaByUserId( + @Param("userId") Long userId, @Param("currentVersion") Long currentVersion) { return "UPDATE " + USER_TABLE_NAME + " SET deleted_at = (UNIX_TIMESTAMP() * 1000.0)" + " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000" - + " WHERE user_id = #{userId} AND deleted_at = 0"; + + " WHERE user_id = #{userId}" + + " AND current_version = #{currentVersion} AND deleted_at = 0"; } public String softDeleteUserMetasByMetalakeId(@Param("metalakeId") Long metalakeId) { @@ -181,11 +180,7 @@ public String updateUserMeta( + " last_version = #{newUserMeta.lastVersion}," + " deleted_at = #{newUserMeta.deletedAt}" + " WHERE user_id = #{oldUserMeta.userId}" - + " AND user_name = #{oldUserMeta.userName}" - + " AND metalake_id = #{oldUserMeta.metalakeId}" - + " AND audit_info = #{oldUserMeta.auditInfo}" + " AND current_version = #{oldUserMeta.currentVersion}" - + " AND last_version = #{oldUserMeta.lastVersion}" + " AND deleted_at = 0"; } 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..462294d882c 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,17 +20,33 @@ 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; public class CatalogMetaPostgreSQLProvider extends CatalogMetaBaseSQLProvider { @Override - public String softDeleteCatalogMetasByCatalogId(Long catalogId) { + public String softDeleteCatalogMetasByCatalogId(Long catalogId, Long currentVersion) { return "UPDATE " + TABLE_NAME + " SET deleted_at = CAST(EXTRACT(EPOCH FROM CURRENT_TIMESTAMP) * 1000 AS BIGINT)" - + " WHERE catalog_id = #{catalogId} AND deleted_at = 0"; + + " WHERE catalog_id = #{catalogId}" + + " AND current_version = #{currentVersion} AND deleted_at = 0"; + } + + /** {@inheritDoc} */ + @Override + public String softDeleteCatalogMetasWithVersion(List catalogPOs) { + return ""; } @Override @@ -101,17 +117,7 @@ public String updateCatalogMeta( + " last_version = #{newCatalogMeta.lastVersion}," + " deleted_at = #{newCatalogMeta.deletedAt}" + " WHERE catalog_id = #{oldCatalogMeta.catalogId}" - + " AND catalog_name = #{oldCatalogMeta.catalogName}" - + " AND metalake_id = #{oldCatalogMeta.metalakeId}" - + " AND type = #{oldCatalogMeta.type}" - + " AND provider = #{oldCatalogMeta.provider}" - + " AND (catalog_comment = #{oldCatalogMeta.catalogComment} " - + " OR (CAST(catalog_comment AS VARCHAR) IS NULL AND " - + " CAST(#{oldCatalogMeta.catalogComment} AS VARCHAR) IS NULL))" - + " AND properties = #{oldCatalogMeta.properties}" - + " AND audit_info = #{oldCatalogMeta.auditInfo}" + " AND current_version = #{oldCatalogMeta.currentVersion}" - + " AND last_version = #{oldCatalogMeta.lastVersion}" + " AND deleted_at = 0"; } } diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/GroupMetaPostgreSQLProvider.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/GroupMetaPostgreSQLProvider.java index 4f617f98b5b..0ba92bfaa48 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/GroupMetaPostgreSQLProvider.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/GroupMetaPostgreSQLProvider.java @@ -29,11 +29,12 @@ public class GroupMetaPostgreSQLProvider extends GroupMetaBaseSQLProvider { @Override - public String softDeleteGroupMetaByGroupId(Long groupId) { + public String softDeleteGroupMetaByGroupId(Long groupId, Long currentVersion) { return "UPDATE " + GROUP_TABLE_NAME + " SET deleted_at = CAST(EXTRACT(EPOCH FROM CURRENT_TIMESTAMP) * 1000 AS BIGINT)" - + " WHERE group_id = #{groupId} AND deleted_at = 0"; + + " WHERE group_id = #{groupId}" + + " AND current_version = #{currentVersion} AND deleted_at = 0"; } @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..b451dd2b8de 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 @@ -98,16 +98,7 @@ public String updateSchemaMeta( + " last_version = #{newSchemaMeta.lastVersion}," + " deleted_at = #{newSchemaMeta.deletedAt}" + " WHERE schema_id = #{oldSchemaMeta.schemaId}" - + " AND schema_name = #{oldSchemaMeta.schemaName}" - + " AND metalake_id = #{oldSchemaMeta.metalakeId}" - + " AND catalog_id = #{oldSchemaMeta.catalogId}" - + " AND (schema_comment = #{oldSchemaMeta.schemaComment}" - + " OR (CAST(schema_comment AS VARCHAR) IS NULL" - + " AND CAST(#{oldSchemaMeta.schemaComment} AS VARCHAR) IS NULL))" - + " AND properties = #{oldSchemaMeta.properties}" - + " AND audit_info = #{oldSchemaMeta.auditInfo}" + " AND current_version = #{oldSchemaMeta.currentVersion}" - + " AND last_version = #{oldSchemaMeta.lastVersion}" + " AND deleted_at = 0"; } @@ -125,6 +116,29 @@ public String softDeleteSchemaMetasBySchemaIds(List schemaIds) { + ""; } + @Override + public String softDeleteSchemaMetaBySchemaIdAndVersion(Long schemaId, Long currentVersion) { + return "UPDATE " + + TABLE_NAME + + " SET deleted_at = CAST(EXTRACT(EPOCH FROM CURRENT_TIMESTAMP) * 1000 AS BIGINT)" + + " WHERE schema_id = #{schemaId}" + + " AND current_version = #{currentVersion} AND deleted_at = 0"; + } + + /** {@inheritDoc} */ + @Override + public String softDeleteSchemaMetasWithVersion(List schemaPOs) { + return ""; + } + @Override public String softDeleteSchemaMetasByMetalakeId(Long metalakeId) { return "UPDATE " diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/UserMetaPostgreSQLProvider.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/UserMetaPostgreSQLProvider.java index 2305535b1d8..8918bbf117a 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/UserMetaPostgreSQLProvider.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/UserMetaPostgreSQLProvider.java @@ -28,11 +28,12 @@ public class UserMetaPostgreSQLProvider extends UserMetaBaseSQLProvider { @Override - public String softDeleteUserMetaByUserId(Long userId) { + public String softDeleteUserMetaByUserId(Long userId, Long currentVersion) { return "UPDATE " + USER_TABLE_NAME + " SET deleted_at = CAST(EXTRACT(EPOCH FROM CURRENT_TIMESTAMP) * 1000 AS BIGINT)" - + " WHERE user_id = #{userId} AND deleted_at = 0"; + + " WHERE user_id = #{userId}" + + " AND current_version = #{currentVersion} AND deleted_at = 0"; } @Override diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/service/CatalogMetaService.java b/core/src/main/java/org/apache/gravitino/storage/relational/service/CatalogMetaService.java index a85e1b85bad..a8886c56c27 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/service/CatalogMetaService.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/service/CatalogMetaService.java @@ -34,8 +34,8 @@ import org.apache.gravitino.Namespace; import org.apache.gravitino.exceptions.NoSuchEntityException; import org.apache.gravitino.exceptions.NonEmptyEntityException; +import org.apache.gravitino.exceptions.OptimisticLockException; import org.apache.gravitino.meta.CatalogEntity; -import org.apache.gravitino.meta.SchemaEntity; import org.apache.gravitino.metrics.Monitored; import org.apache.gravitino.storage.relational.helper.CatalogIds; import org.apache.gravitino.storage.relational.mapper.CatalogMetaMapper; @@ -44,6 +44,7 @@ import org.apache.gravitino.storage.relational.mapper.FilesetVersionMapper; import org.apache.gravitino.storage.relational.mapper.FunctionMetaMapper; import org.apache.gravitino.storage.relational.mapper.FunctionVersionMetaMapper; +import org.apache.gravitino.storage.relational.mapper.MetalakeMetaMapper; import org.apache.gravitino.storage.relational.mapper.ModelMetaMapper; import org.apache.gravitino.storage.relational.mapper.ModelVersionAliasRelMapper; import org.apache.gravitino.storage.relational.mapper.ModelVersionMetaMapper; @@ -58,6 +59,8 @@ import org.apache.gravitino.storage.relational.mapper.TopicMetaMapper; 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.po.cache.OperateType; import org.apache.gravitino.storage.relational.utils.ExceptionUtils; import org.apache.gravitino.storage.relational.utils.POConverters; @@ -181,20 +184,32 @@ public void insertCatalog(CatalogEntity catalogEntity, boolean overwrite) throws try { NameIdentifierUtil.checkCatalog(catalogEntity.nameIdentifier()); - String metalake = NameIdentifierUtil.getMetalake(catalogEntity.nameIdentifier()); - Long metalakeId = - EntityIdService.getEntityId(NameIdentifier.of(metalake), Entity.EntityType.METALAKE); - - SessionUtils.doWithCommit( - CatalogMetaMapper.class, - mapper -> { - CatalogPO po = POConverters.initializeCatalogPOWithVersion(catalogEntity, metalakeId); - if (overwrite) { - mapper.insertCatalogMetaOnDuplicateKeyUpdate(po); - } else { - mapper.insertCatalogMeta(po); - } - }); + String metalakeName = NameIdentifierUtil.getMetalake(catalogEntity.nameIdentifier()); + MetalakePO metalakePO = + SessionUtils.getWithoutCommit( + MetalakeMetaMapper.class, mapper -> mapper.selectMetalakeMetaByName(metalakeName)); + if (metalakePO == null) { + throw new NoSuchEntityException( + NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE, + Entity.EntityType.METALAKE.name().toLowerCase(), + metalakeName); + } + + SessionUtils.doMultipleWithCommit( + () -> fenceMetalakeForCatalogCreate(metalakePO), + () -> + SessionUtils.doWithoutCommit( + CatalogMetaMapper.class, + mapper -> { + CatalogPO po = + POConverters.initializeCatalogPOWithVersion( + catalogEntity, metalakePO.getMetalakeId()); + if (overwrite) { + mapper.insertCatalogMetaOnDuplicateKeyUpdate(po); + } else { + mapper.insertCatalogMeta(po); + } + })); } catch (RuntimeException re) { ExceptionUtils.checkSQLException( re, Entity.EntityType.CATALOG, catalogEntity.nameIdentifier().toString()); @@ -229,15 +244,19 @@ public CatalogEntity updateCatalog( AtomicInteger updateResult = new AtomicInteger(0); try { SessionUtils.doMultipleWithCommit( - () -> - updateResult.set( - SessionUtils.getWithoutCommit( - CatalogMetaMapper.class, - mapper -> - mapper.updateCatalogMeta( - POConverters.updateCatalogPOWithVersion( - oldCatalogPO, newEntity, oldCatalogPO.getMetalakeId()), - oldCatalogPO))), + () -> { + updateResult.set( + SessionUtils.getWithoutCommit( + CatalogMetaMapper.class, + mapper -> + mapper.updateCatalogMeta( + POConverters.updateCatalogPOWithVersion( + oldCatalogPO, newEntity, oldCatalogPO.getMetalakeId()), + oldCatalogPO))); + if (updateResult.get() == 0) { + throw optimisticLockException(identifier); + } + }, () -> { if (updateResult.get() > 0) { SessionUtils.doWithoutCommit( @@ -256,11 +275,7 @@ public CatalogEntity updateCatalog( throw re; } - if (updateResult.get() > 0) { - return newEntity; - } else { - throw new IOException("Failed to update the entity: " + identifier); - } + return newEntity; } @Monitored( @@ -270,19 +285,17 @@ public boolean deleteCatalog(NameIdentifier identifier, boolean cascade) { NameIdentifierUtil.checkCatalog(identifier); String catalogName = identifier.name(); - long catalogId = EntityIdService.getEntityId(identifier, Entity.EntityType.CATALOG); + CatalogPO catalogPO = getCatalogPOByName(identifier.namespace().level(0), catalogName); + long catalogId = catalogPO.getCatalogId(); + long currentVersion = catalogPO.getCurrentVersion(); String metalakeName = identifier.namespace().level(0); if (cascade) { SessionUtils.doMultipleWithCommit( - () -> - SessionUtils.doWithoutCommit( - CatalogMetaMapper.class, - mapper -> mapper.softDeleteCatalogMetasByCatalogId(catalogId)), - () -> - SessionUtils.doWithoutCommit( - SchemaMetaMapper.class, - mapper -> mapper.softDeleteSchemaMetasByCatalogId(catalogId)), + () -> { + deleteCatalogWithVersion(identifier, catalogId, currentVersion); + deleteSchemasWithVersions(identifier, catalogId); + }, () -> SessionUtils.doWithoutCommit( TableMetaMapper.class, @@ -359,19 +372,17 @@ public boolean deleteCatalog(NameIdentifier identifier, boolean cascade) { OperateType.DROP)); }); } else { - List schemaEntities = - SchemaMetaService.getInstance() - .listSchemasByNamespace( - NamespaceUtil.ofSchema(identifier.namespace().level(0), catalogName)); - if (!schemaEntities.isEmpty()) { - throw new NonEmptyEntityException( - "Entity %s has sub-entities, you should remove sub-entities first", identifier); - } SessionUtils.doMultipleWithCommit( - () -> - SessionUtils.doWithoutCommit( - CatalogMetaMapper.class, - mapper -> mapper.softDeleteCatalogMetasByCatalogId(catalogId)), + () -> { + deleteCatalogWithVersion(identifier, catalogId, currentVersion); + List schemaPOs = + SessionUtils.getWithoutCommit( + SchemaMetaMapper.class, mapper -> mapper.listSchemaPOsByCatalogId(catalogId)); + if (!schemaPOs.isEmpty()) { + throw new NonEmptyEntityException( + "Entity %s has sub-entities, you should remove sub-entities first", identifier); + } + }, () -> SessionUtils.doWithoutCommit( OwnerMetaMapper.class, @@ -415,6 +426,53 @@ public boolean deleteCatalog(NameIdentifier identifier, boolean cascade) { return true; } + private void deleteCatalogWithVersion( + NameIdentifier identifier, Long catalogId, Long currentVersion) { + int deleted = + SessionUtils.getWithoutCommit( + CatalogMetaMapper.class, + mapper -> mapper.softDeleteCatalogMetasByCatalogId(catalogId, currentVersion)); + if (deleted == 0) { + throw optimisticLockException(identifier); + } + } + + private void fenceMetalakeForCatalogCreate(MetalakePO metalakePO) { + int fenced = + SessionUtils.getWithoutCommit( + MetalakeMetaMapper.class, + mapper -> + mapper.fenceMetalakeMeta( + metalakePO.getMetalakeId(), metalakePO.getCurrentVersion())); + if (fenced == 0) { + throw new OptimisticLockException( + "The parent metalake %s was modified concurrently; retry the operation", + metalakePO.getMetalakeName()); + } + } + + private void deleteSchemasWithVersions(NameIdentifier catalogIdentifier, Long catalogId) { + List schemaPOs = + SessionUtils.getWithoutCommit( + SchemaMetaMapper.class, mapper -> mapper.listSchemaPOsByCatalogId(catalogId)); + if (schemaPOs.isEmpty()) { + return; + } + int deleted = + SessionUtils.getWithoutCommit( + SchemaMetaMapper.class, mapper -> mapper.softDeleteSchemaMetasWithVersion(schemaPOs)); + if (deleted != schemaPOs.size()) { + throw new OptimisticLockException( + "A schema under catalog %s was modified concurrently; retry the operation", + catalogIdentifier); + } + } + + private OptimisticLockException optimisticLockException(NameIdentifier identifier) { + return new OptimisticLockException( + "The catalog %s was modified concurrently; retry the operation", identifier); + } + @Monitored( metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME, baseMetricName = "deleteCatalogMetasByLegacyTimeline") diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/service/GroupMetaService.java b/core/src/main/java/org/apache/gravitino/storage/relational/service/GroupMetaService.java index 50b20038626..b29fcfc61e8 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/service/GroupMetaService.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/service/GroupMetaService.java @@ -37,15 +37,18 @@ import org.apache.gravitino.Namespace; import org.apache.gravitino.authorization.AuthorizationUtils; import org.apache.gravitino.exceptions.NoSuchEntityException; +import org.apache.gravitino.exceptions.OptimisticLockException; import org.apache.gravitino.meta.GroupEntity; import org.apache.gravitino.meta.RoleEntity; import org.apache.gravitino.metrics.Monitored; import org.apache.gravitino.storage.relational.mapper.GroupMetaMapper; import org.apache.gravitino.storage.relational.mapper.GroupRoleRelMapper; +import org.apache.gravitino.storage.relational.mapper.MetalakeMetaMapper; import org.apache.gravitino.storage.relational.mapper.OwnerMetaMapper; import org.apache.gravitino.storage.relational.po.ExtendedGroupPO; import org.apache.gravitino.storage.relational.po.GroupPO; import org.apache.gravitino.storage.relational.po.GroupRoleRelPO; +import org.apache.gravitino.storage.relational.po.MetalakePO; import org.apache.gravitino.storage.relational.po.RolePO; import org.apache.gravitino.storage.relational.utils.ExceptionUtils; import org.apache.gravitino.storage.relational.utils.POConverters; @@ -171,9 +174,18 @@ public void insertGroup(GroupEntity groupEntity, boolean overwritten) throws IOE NameIdentifier metalakeIdent = NameIdentifier.of(NameIdentifierUtil.getMetalake(groupEntity.nameIdentifier())); - Long metalakeId = EntityIdService.getEntityId(metalakeIdent, Entity.EntityType.METALAKE); - - GroupPO.Builder builder = GroupPO.builder().withMetalakeId(metalakeId); + MetalakePO metalakePO = + SessionUtils.getWithoutCommit( + MetalakeMetaMapper.class, + mapper -> mapper.selectMetalakeMetaByName(metalakeIdent.name())); + if (metalakePO == null) { + throw new NoSuchEntityException( + NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE, + Entity.EntityType.METALAKE.name().toLowerCase(), + metalakeIdent.name()); + } + + GroupPO.Builder builder = GroupPO.builder().withMetalakeId(metalakePO.getMetalakeId()); GroupPO GroupPO = POConverters.initializeGroupPOWithVersion(groupEntity, builder); List roleIds = Optional.ofNullable(groupEntity.roleIds()).orElse(Lists.newArrayList()); @@ -181,6 +193,7 @@ public void insertGroup(GroupEntity groupEntity, boolean overwritten) throws IOE POConverters.initializeGroupRoleRelsPOWithVersion(groupEntity, roleIds); SessionUtils.doMultipleWithCommit( + () -> fenceMetalakeForGroupCreate(metalakePO), () -> SessionUtils.doWithoutCommit( GroupMetaMapper.class, @@ -214,12 +227,22 @@ public void insertGroup(GroupEntity groupEntity, boolean overwritten) throws IOE public boolean deleteGroup(NameIdentifier identifier) { AuthorizationUtils.checkGroup(identifier); - Long groupId = EntityIdService.getEntityId(identifier, Entity.EntityType.GROUP); + Long metalakeId = + MetalakeMetaService.getInstance().getMetalakeIdByName(identifier.namespace().level(0)); + GroupPO groupPO = getGroupPOByMetalakeIdAndName(metalakeId, identifier.name()); + Long groupId = groupPO.getGroupId(); SessionUtils.doMultipleWithCommit( - () -> - SessionUtils.doWithoutCommit( - GroupMetaMapper.class, mapper -> mapper.softDeleteGroupMetaByGroupId(groupId)), + () -> { + int deleted = + SessionUtils.getWithoutCommit( + GroupMetaMapper.class, + mapper -> + mapper.softDeleteGroupMetaByGroupId(groupId, groupPO.getCurrentVersion())); + if (deleted == 0) { + throw optimisticLockException(identifier); + } + }, () -> SessionUtils.doWithoutCommit( GroupRoleRelMapper.class, @@ -264,18 +287,20 @@ public GroupEntity updateGroup( Set insertRoleIds = Sets.difference(newRoleIds, oldRoleIds); Set deleteRoleIds = Sets.difference(oldRoleIds, newRoleIds); - if (insertRoleIds.isEmpty() && deleteRoleIds.isEmpty()) { - return newEntity; - } try { SessionUtils.doMultipleWithCommit( - () -> - SessionUtils.doWithoutCommit( - GroupMetaMapper.class, - mapper -> - mapper.updateGroupMeta( - POConverters.updateGroupPOWithVersion(oldGroupPO, newEntity), - oldGroupPO)), + () -> { + int updated = + SessionUtils.getWithoutCommit( + GroupMetaMapper.class, + mapper -> + mapper.updateGroupMeta( + POConverters.updateGroupPOWithVersion(oldGroupPO, newEntity), + oldGroupPO)); + if (updated == 0) { + throw optimisticLockException(identifier); + } + }, () -> { if (insertRoleIds.isEmpty()) { return; @@ -394,21 +419,6 @@ public GroupEntity getGroupByExternalId(NameIdentifier ident) { groupPO, rolePOs, AuthorizationUtils.ofGroupNamespace(metalake)); } - private GroupPO getGroupPOByMetalakeNameAndId(String metalakeName, Long groupId) { - GroupPO groupPO = - SessionUtils.getWithoutCommit( - GroupMetaMapper.class, - mapper -> mapper.selectGroupMetaByMetalakeNameAndId(metalakeName, groupId)); - - if (groupPO == null) { - throw new NoSuchEntityException( - NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE, - Entity.EntityType.GROUP.name().toLowerCase(), - String.valueOf(groupId)); - } - return groupPO; - } - @Monitored( metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME, baseMetricName = "getGroupById") @@ -439,13 +449,18 @@ public GroupEntity updateGroupById( try { SessionUtils.doMultipleWithCommit( - () -> - SessionUtils.doWithoutCommit( - GroupMetaMapper.class, - mapper -> - mapper.updateGroupMeta( - POConverters.updateGroupPOWithVersion(oldGroupPO, newEntity), - oldGroupPO)), + () -> { + int updated = + SessionUtils.getWithoutCommit( + GroupMetaMapper.class, + mapper -> + mapper.updateGroupMeta( + POConverters.updateGroupPOWithVersion(oldGroupPO, newEntity), + oldGroupPO)); + if (updated == 0) { + throw optimisticLockException(newEntity.nameIdentifier()); + } + }, () -> SessionUtils.doWithoutCommit( GroupMetaMapper.class, @@ -462,16 +477,25 @@ public GroupEntity updateGroupById( metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME, baseMetricName = "deleteGroupById") public boolean deleteGroupById(String metalake, long groupId) { + GroupPO groupPO; try { - getGroupPOByMetalakeNameAndId(metalake, groupId); + groupPO = getGroupPOByMetalakeNameAndId(metalake, groupId); } catch (NoSuchEntityException e) { return false; } + NameIdentifier identifier = AuthorizationUtils.ofGroup(metalake, groupPO.getGroupName()); SessionUtils.doMultipleWithCommit( - () -> - SessionUtils.doWithoutCommit( - GroupMetaMapper.class, mapper -> mapper.softDeleteGroupMetaByGroupId(groupId)), + () -> { + int deleted = + SessionUtils.getWithoutCommit( + GroupMetaMapper.class, + mapper -> + mapper.softDeleteGroupMetaByGroupId(groupId, groupPO.getCurrentVersion())); + if (deleted == 0) { + throw optimisticLockException(identifier); + } + }, () -> SessionUtils.doWithoutCommit( GroupRoleRelMapper.class, @@ -484,4 +508,38 @@ public boolean deleteGroupById(String metalake, long groupId) { groupId, Entity.EntityType.GROUP.name()))); return true; } + + private GroupPO getGroupPOByMetalakeNameAndId(String metalakeName, Long groupId) { + GroupPO groupPO = + SessionUtils.getWithoutCommit( + GroupMetaMapper.class, + mapper -> mapper.selectGroupMetaByMetalakeNameAndId(metalakeName, groupId)); + + if (groupPO == null) { + throw new NoSuchEntityException( + NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE, + Entity.EntityType.GROUP.name().toLowerCase(), + String.valueOf(groupId)); + } + return groupPO; + } + + private OptimisticLockException optimisticLockException(NameIdentifier identifier) { + return new OptimisticLockException( + "The group %s was modified concurrently; retry the operation", identifier); + } + + private void fenceMetalakeForGroupCreate(MetalakePO metalakePO) { + int fenced = + SessionUtils.getWithoutCommit( + MetalakeMetaMapper.class, + mapper -> + mapper.fenceMetalakeMeta( + metalakePO.getMetalakeId(), metalakePO.getCurrentVersion())); + if (fenced == 0) { + throw new OptimisticLockException( + "The parent metalake %s was modified concurrently; retry the operation", + metalakePO.getMetalakeName()); + } + } } 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 607dab00c11..824279fa143 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 @@ -33,8 +33,8 @@ import org.apache.gravitino.NameIdentifier; 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.CatalogEntity; import org.apache.gravitino.metrics.Monitored; import org.apache.gravitino.storage.relational.mapper.CatalogMetaMapper; import org.apache.gravitino.storage.relational.mapper.EntityChangeLogMapper; @@ -65,13 +65,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.cache.OperateType; 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. @@ -183,11 +183,15 @@ public BaseMetalake updateMetalake( AtomicInteger updateResult = new AtomicInteger(0); try { SessionUtils.doMultipleWithCommit( - () -> - updateResult.set( - SessionUtils.getWithoutCommit( - MetalakeMetaMapper.class, - mapper -> mapper.updateMetalakeMeta(newMetalakePO, oldMetalakePO))), + () -> { + updateResult.set( + SessionUtils.getWithoutCommit( + MetalakeMetaMapper.class, + mapper -> mapper.updateMetalakeMeta(newMetalakePO, oldMetalakePO))); + if (updateResult.get() == 0) { + throw optimisticLockException(ident); + } + }, () -> { if (isRenamed && updateResult.get() > 0) { SessionUtils.doWithoutCommit( @@ -206,11 +210,7 @@ public BaseMetalake updateMetalake( throw re; } - if (updateResult.get() > 0) { - return newMetalakeEntity; - } else { - throw new IOException("Failed to update the entity: " + ident); - } + return newMetalakeEntity; } @Monitored( @@ -218,18 +218,24 @@ 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)), + () -> { + deleteMetalakeWithVersion(ident, metalakeId, currentVersion); + deleteCatalogsWithVersions(ident, metalakeId); + }, () -> SessionUtils.doWithoutCommit( SchemaMetaMapper.class, @@ -345,18 +351,18 @@ public boolean deleteMetalake(NameIdentifier ident, boolean cascade) { OperateType.DROP)); }); } 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)), + () -> { + 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, @@ -420,6 +426,39 @@ 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 optimisticLockException(identifier); + } + } + + private void deleteCatalogsWithVersions(NameIdentifier metalakeIdentifier, Long metalakeId) { + List catalogPOs = + SessionUtils.getWithoutCommit( + CatalogMetaMapper.class, mapper -> mapper.listCatalogPOsByMetalakeId(metalakeId)); + if (catalogPOs.isEmpty()) { + return; + } + int deleted = + SessionUtils.getWithoutCommit( + CatalogMetaMapper.class, + mapper -> mapper.softDeleteCatalogMetasWithVersion(catalogPOs)); + if (deleted != catalogPOs.size()) { + throw new OptimisticLockException( + "A catalog under metalake %s was modified concurrently; retry the operation", + metalakeIdentifier); + } + } + + private OptimisticLockException optimisticLockException(NameIdentifier identifier) { + return new OptimisticLockException( + "The metalake %s was modified concurrently; retry the operation", identifier); + } + @Monitored( metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME, baseMetricName = "deleteMetalakeMetasByLegacyTimeline") diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/service/SchemaMetaService.java b/core/src/main/java/org/apache/gravitino/storage/relational/service/SchemaMetaService.java index 39d820b42d0..f69ee3e96a5 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/service/SchemaMetaService.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/service/SchemaMetaService.java @@ -26,9 +26,10 @@ import java.util.Arrays; import java.util.Collections; import java.util.List; +import java.util.Map; import java.util.Objects; -import java.util.Set; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import java.util.function.Function; import java.util.regex.Pattern; import java.util.stream.Collectors; @@ -40,15 +41,12 @@ import org.apache.gravitino.Namespace; import org.apache.gravitino.exceptions.NoSuchEntityException; import org.apache.gravitino.exceptions.NonEmptyEntityException; -import org.apache.gravitino.meta.FilesetEntity; -import org.apache.gravitino.meta.ModelEntity; -import org.apache.gravitino.meta.NamespacedEntityId; +import org.apache.gravitino.exceptions.OptimisticLockException; import org.apache.gravitino.meta.SchemaEntity; -import org.apache.gravitino.meta.TableEntity; -import org.apache.gravitino.meta.TopicEntity; import org.apache.gravitino.metrics.Monitored; import org.apache.gravitino.storage.IdGenerator; import org.apache.gravitino.storage.relational.helper.SchemaIds; +import org.apache.gravitino.storage.relational.mapper.CatalogMetaMapper; import org.apache.gravitino.storage.relational.mapper.EntityChangeLogMapper; import org.apache.gravitino.storage.relational.mapper.FilesetMetaMapper; import org.apache.gravitino.storage.relational.mapper.FilesetVersionMapper; @@ -67,6 +65,7 @@ import org.apache.gravitino.storage.relational.mapper.TagMetadataObjectRelMapper; import org.apache.gravitino.storage.relational.mapper.TopicMetaMapper; import org.apache.gravitino.storage.relational.mapper.ViewMetaMapper; +import org.apache.gravitino.storage.relational.po.CatalogPO; import org.apache.gravitino.storage.relational.po.SchemaPO; import org.apache.gravitino.storage.relational.po.cache.OperateType; import org.apache.gravitino.storage.relational.utils.ExceptionUtils; @@ -145,6 +144,10 @@ public void insertSchema(SchemaEntity schemaEntity, boolean overwrite) throws IO // rewriter to translate each PO's name to storage form before SQL execution. String logicalSep = HierarchicalSchemaUtil.schemaSeparator(); String schemaName = schemaEntity.name(); + String metalakeName = schemaEntity.namespace().level(0); + String catalogName = schemaEntity.namespace().level(1); + CatalogPO catalogPO = + CatalogMetaService.getInstance().getCatalogPOByName(metalakeName, catalogName); List rowsToInsert = new ArrayList<>(); if (schemaName == null || !schemaName.contains(logicalSep)) { rowsToInsert.add(schemaEntity); @@ -167,39 +170,46 @@ public void insertSchema(SchemaEntity schemaEntity, boolean overwrite) throws IO rowsToInsert.add(schemaEntity); } - SessionUtils.doWithCommit( - SchemaMetaMapper.class, - mapper -> { - int n = rowsToInsert.size(); - List missingAncestorPOs = new ArrayList<>(); - if (n > 1) { - SchemaEntity firstAncestor = rowsToInsert.get(0); - Namespace ancestorNs = firstAncestor.namespace(); - List ancestorNames = - rowsToInsert.subList(0, n - 1).stream() - .map(SchemaEntity::name) - .collect(Collectors.toList()); - Set existingLogicalNames = - ops.listPOs(mapper, ancestorNs, ancestorNames).stream() - .map(SchemaPO::getSchemaName) - .collect(Collectors.toSet()); - for (SchemaEntity row : rowsToInsert.subList(0, n - 1)) { - if (existingLogicalNames.contains(row.name())) { - continue; - } - SchemaPO.Builder builder = SchemaPO.builder(); - fillSchemaPOBuilderParentEntityId(builder, row.namespace()); - missingAncestorPOs.add(POConverters.initializeSchemaPOWithVersion(row, builder)); - } - } - SchemaEntity leafRow = rowsToInsert.get(n - 1); - SchemaPO.Builder leafBuilder = SchemaPO.builder(); - fillSchemaPOBuilderParentEntityId(leafBuilder, leafRow.namespace()); - SchemaPO leafPO = POConverters.initializeSchemaPOWithVersion(leafRow, leafBuilder); - List schemaPosToInsert = new ArrayList<>(missingAncestorPOs); - schemaPosToInsert.add(leafPO); - ops.batchInsertPOs(mapper, schemaPosToInsert, overwrite); - }); + SessionUtils.doMultipleWithCommit( + () -> fenceCatalogForSchemaCreate(catalogPO), + () -> + SessionUtils.doWithoutCommit( + SchemaMetaMapper.class, + mapper -> { + int n = rowsToInsert.size(); + List missingAncestorPOs = new ArrayList<>(); + if (n > 1) { + SchemaEntity firstAncestor = rowsToInsert.get(0); + Namespace ancestorNs = firstAncestor.namespace(); + List ancestorNames = + rowsToInsert.subList(0, n - 1).stream() + .map(SchemaEntity::name) + .collect(Collectors.toList()); + Map existingAncestors = + ops.listPOs(mapper, ancestorNs, ancestorNames).stream() + .collect( + Collectors.toMap(SchemaPO::getSchemaName, Function.identity())); + for (SchemaEntity row : rowsToInsert.subList(0, n - 1)) { + SchemaPO existingAncestor = existingAncestors.get(row.name()); + if (existingAncestor != null) { + fenceSchemaAncestor( + mapper, existingAncestor, schemaEntity.nameIdentifier()); + continue; + } + SchemaPO.Builder builder = newSchemaPOBuilder(catalogPO); + missingAncestorPOs.add( + POConverters.initializeSchemaPOWithVersion(row, builder)); + } + } + if (!missingAncestorPOs.isEmpty()) { + ops.batchInsertPOs(mapper, missingAncestorPOs, false); + } + SchemaEntity leafRow = rowsToInsert.get(n - 1); + SchemaPO leafPO = + POConverters.initializeSchemaPOWithVersion( + leafRow, newSchemaPOBuilder(catalogPO)); + ops.batchInsertPOs(mapper, Collections.singletonList(leafPO), overwrite); + })); } catch (RuntimeException re) { ExceptionUtils.checkSQLException( re, Entity.EntityType.SCHEMA, schemaEntity.nameIdentifier().toString()); @@ -230,15 +240,19 @@ public SchemaEntity updateSchema( AtomicInteger updateResult = new AtomicInteger(0); try { SessionUtils.doMultipleWithCommit( - () -> - updateResult.set( - SessionUtils.getWithoutCommit( - SchemaMetaMapper.class, - mapper -> - ops.updatePO( - mapper, - POConverters.updateSchemaPOWithVersion(oldSchemaPO, newEntity), - oldSchemaPO))), + () -> { + updateResult.set( + SessionUtils.getWithoutCommit( + SchemaMetaMapper.class, + mapper -> + ops.updatePO( + mapper, + POConverters.updateSchemaPOWithVersion(oldSchemaPO, newEntity), + oldSchemaPO))); + if (updateResult.get() == 0) { + throw optimisticLockException(identifier); + } + }, () -> { if (isRenamed && updateResult.get() > 0) { SessionUtils.doWithoutCommit( @@ -257,11 +271,7 @@ public SchemaEntity updateSchema( throw re; } - if (updateResult.get() > 0) { - return newEntity; - } else { - throw new IOException("Failed to update the entity: " + identifier); - } + return newEntity; } @Monitored( @@ -279,81 +289,81 @@ public boolean deleteSchema(NameIdentifier identifier, boolean cascade) { NameIdentifierUtil.ofSchema(metalakeName, catalogName, schemaName).toString(); if (cascade) { - // For HierarchicalSchema, deleting `A:B` must also cascade into all descendant schemas - // such as `A:B:C`, `A:B:C:D`, etc. Collect the descendant schema ids up-front and run a - // single batch UPDATE per child table so the total SQL cost stays bounded regardless of - // how many descendants exist. - List schemaIds = listSchemaIdsForCascade(schemaPO); - if (schemaIds.isEmpty()) { - return false; - } + AtomicReference> schemaIds = new AtomicReference<>(); SessionUtils.doMultipleWithCommit( - () -> - SessionUtils.doWithoutCommit( - SchemaMetaMapper.class, - mapper -> mapper.softDeleteSchemaMetasBySchemaIds(schemaIds)), + () -> { + deleteSchemaWithVersion(identifier, schemaId, schemaPO.getCurrentVersion()); + List descendants = listDescendantSchemaPOs(schemaPO); + deleteDescendantSchemasWithVersions(identifier, descendants); + List ids = new ArrayList<>(descendants.size() + 1); + ids.add(schemaId); + descendants.stream().map(SchemaPO::getSchemaId).forEach(ids::add); + schemaIds.set(ids); + }, () -> SessionUtils.doWithoutCommit( TableMetaMapper.class, - mapper -> mapper.softDeleteTableMetasBySchemaIds(schemaIds)), + mapper -> mapper.softDeleteTableMetasBySchemaIds(schemaIds.get())), () -> SessionUtils.doWithoutCommit( TableColumnMapper.class, - mapper -> mapper.softDeleteColumnsBySchemaIds(schemaIds)), + mapper -> mapper.softDeleteColumnsBySchemaIds(schemaIds.get())), () -> SessionUtils.doWithoutCommit( FilesetMetaMapper.class, - mapper -> mapper.softDeleteFilesetMetasBySchemaIds(schemaIds)), + mapper -> mapper.softDeleteFilesetMetasBySchemaIds(schemaIds.get())), () -> SessionUtils.doWithoutCommit( FilesetVersionMapper.class, - mapper -> mapper.softDeleteFilesetVersionsBySchemaIds(schemaIds)), + mapper -> mapper.softDeleteFilesetVersionsBySchemaIds(schemaIds.get())), () -> SessionUtils.doWithoutCommit( TopicMetaMapper.class, - mapper -> mapper.softDeleteTopicMetasBySchemaIds(schemaIds)), + mapper -> mapper.softDeleteTopicMetasBySchemaIds(schemaIds.get())), () -> SessionUtils.doWithoutCommit( FunctionMetaMapper.class, - mapper -> mapper.softDeleteFunctionMetasBySchemaIds(schemaIds)), + mapper -> mapper.softDeleteFunctionMetasBySchemaIds(schemaIds.get())), () -> SessionUtils.doWithoutCommit( FunctionVersionMetaMapper.class, - mapper -> mapper.softDeleteFunctionVersionMetasBySchemaIds(schemaIds)), + mapper -> mapper.softDeleteFunctionVersionMetasBySchemaIds(schemaIds.get())), () -> SessionUtils.doWithoutCommit( - OwnerMetaMapper.class, mapper -> mapper.softDeleteOwnerRelBySchemaIds(schemaIds)), + OwnerMetaMapper.class, + mapper -> mapper.softDeleteOwnerRelBySchemaIds(schemaIds.get())), () -> SessionUtils.doWithoutCommit( SecurableObjectMapper.class, - mapper -> mapper.softDeleteObjectRelsBySchemaIds(schemaIds)), + mapper -> mapper.softDeleteObjectRelsBySchemaIds(schemaIds.get())), () -> SessionUtils.doWithoutCommit( TagMetadataObjectRelMapper.class, - mapper -> mapper.softDeleteTagMetadataObjectRelsBySchemaIds(schemaIds)), + mapper -> mapper.softDeleteTagMetadataObjectRelsBySchemaIds(schemaIds.get())), () -> SessionUtils.doWithoutCommit( PolicyMetadataObjectRelMapper.class, - mapper -> mapper.softDeletePolicyMetadataObjectRelsBySchemaIds(schemaIds)), + mapper -> mapper.softDeletePolicyMetadataObjectRelsBySchemaIds(schemaIds.get())), () -> SessionUtils.doWithoutCommit( ModelVersionAliasRelMapper.class, - mapper -> mapper.softDeleteModelVersionAliasRelsBySchemaIds(schemaIds)), + mapper -> mapper.softDeleteModelVersionAliasRelsBySchemaIds(schemaIds.get())), () -> SessionUtils.doWithoutCommit( ModelVersionMetaMapper.class, - mapper -> mapper.softDeleteModelVersionMetasBySchemaIds(schemaIds)), + mapper -> mapper.softDeleteModelVersionMetasBySchemaIds(schemaIds.get())), () -> SessionUtils.doWithoutCommit( ModelMetaMapper.class, - mapper -> mapper.softDeleteModelMetasBySchemaIds(schemaIds)), + mapper -> mapper.softDeleteModelMetasBySchemaIds(schemaIds.get())), () -> SessionUtils.doWithoutCommit( StatisticMetaMapper.class, - mapper -> mapper.softDeleteStatisticsBySchemaIds(schemaIds)), + mapper -> mapper.softDeleteStatisticsBySchemaIds(schemaIds.get())), () -> SessionUtils.doWithoutCommit( - ViewMetaMapper.class, mapper -> mapper.softDeleteViewMetasBySchemaIds(schemaIds)), + ViewMetaMapper.class, + mapper -> mapper.softDeleteViewMetasBySchemaIds(schemaIds.get())), () -> { SessionUtils.doWithoutCommit( EntityChangeLogMapper.class, @@ -365,58 +375,11 @@ public boolean deleteSchema(NameIdentifier identifier, boolean cascade) { OperateType.DROP)); }); } else { - List tableEntities = - TableMetaService.getInstance() - .listTablesByNamespace( - NamespaceUtil.ofTable( - identifier.namespace().level(0), - identifier.namespace().level(1), - schemaName)); - if (!tableEntities.isEmpty()) { - throw new NonEmptyEntityException( - "Entity %s has sub-entities, you should remove sub-entities first", identifier); - } - List filesetEntities = - FilesetMetaService.getInstance() - .listFilesetsByNamespace( - NamespaceUtil.ofFileset( - identifier.namespace().level(0), - identifier.namespace().level(1), - schemaName)); - if (!filesetEntities.isEmpty()) { - throw new NonEmptyEntityException( - "Entity %s has sub-entities, you should remove sub-entities first", identifier); - } - List modelEntities = - ModelMetaService.getInstance() - .listModelsByNamespace( - NamespaceUtil.ofModel( - identifier.namespace().level(0), - identifier.namespace().level(1), - schemaName)); - if (!modelEntities.isEmpty()) { - throw new NonEmptyEntityException( - "Entity %s has sub-entities, you should remove sub-entities first", identifier); - } - - List topicEntities = - TopicMetaService.getInstance() - .listTopicsByNamespace( - NamespaceUtil.ofTopic( - identifier.namespace().level(0), - identifier.namespace().level(1), - schemaName)); - if (!topicEntities.isEmpty()) { - throw new NonEmptyEntityException( - "Entity %s has sub-entities, you should remove sub-entities first", identifier); - } - - List singleSchemaId = Collections.singletonList(schemaId); SessionUtils.doMultipleWithCommit( - () -> - SessionUtils.doWithoutCommit( - SchemaMetaMapper.class, - mapper -> mapper.softDeleteSchemaMetasBySchemaIds(singleSchemaId)), + () -> { + deleteSchemaWithVersion(identifier, schemaId, schemaPO.getCurrentVersion()); + checkSchemaIsEmpty(identifier, schemaPO); + }, () -> SessionUtils.doWithoutCommit( OwnerMetaMapper.class, @@ -459,6 +422,22 @@ public boolean deleteSchema(NameIdentifier identifier, boolean cascade) { return true; } + private void deleteSchemaWithVersion( + NameIdentifier identifier, Long schemaId, Long currentVersion) { + int deleted = + SessionUtils.getWithoutCommit( + SchemaMetaMapper.class, + mapper -> mapper.softDeleteSchemaMetaBySchemaIdAndVersion(schemaId, currentVersion)); + if (deleted == 0) { + throw optimisticLockException(identifier); + } + } + + private OptimisticLockException optimisticLockException(NameIdentifier identifier) { + return new OptimisticLockException( + "The schema %s was modified concurrently; retry the operation", identifier); + } + @Monitored( metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME, baseMetricName = "deleteSchemaMetasByLegacyTimeline") @@ -492,13 +471,78 @@ private List listSchemaPOs(Namespace namespace) { mapper -> POStorageReadRouting.listPOs(mapper, namespace, ops, Entity.EntityType.SCHEMA)); } + private void fenceCatalogForSchemaCreate(CatalogPO catalogPO) { + int fenced = + SessionUtils.getWithoutCommit( + CatalogMetaMapper.class, + mapper -> + mapper.fenceCatalogMeta(catalogPO.getCatalogId(), catalogPO.getCurrentVersion())); + if (fenced == 0) { + throw new OptimisticLockException( + "The parent catalog %s was modified concurrently; retry the operation", + catalogPO.getCatalogName()); + } + } + + private void fenceSchemaAncestor( + SchemaMetaMapper mapper, SchemaPO ancestor, NameIdentifier schemaIdentifier) { + int fenced = mapper.fenceSchemaMeta(ancestor.getSchemaId(), ancestor.getCurrentVersion()); + if (fenced == 0) { + throw new OptimisticLockException( + "An ancestor of schema %s was modified concurrently; retry the operation", + schemaIdentifier); + } + } + + private void deleteDescendantSchemasWithVersions( + NameIdentifier schemaIdentifier, List descendants) { + if (descendants.isEmpty()) { + return; + } + int deleted = + SessionUtils.getWithoutCommit( + SchemaMetaMapper.class, mapper -> mapper.softDeleteSchemaMetasWithVersion(descendants)); + if (deleted != descendants.size()) { + throw new OptimisticLockException( + "A descendant of schema %s was modified concurrently; retry the operation", + schemaIdentifier); + } + } + + private void checkSchemaIsEmpty(NameIdentifier identifier, SchemaPO schemaPO) { + boolean hasDescendantSchemas = !listDescendantSchemaPOs(schemaPO).isEmpty(); + boolean hasTables = + !SessionUtils.getWithoutCommit( + TableMetaMapper.class, + mapper -> mapper.listTablePOsBySchemaId(schemaPO.getSchemaId())) + .isEmpty(); + boolean hasFilesets = + !SessionUtils.getWithoutCommit( + FilesetMetaMapper.class, + mapper -> mapper.listFilesetPOsBySchemaId(schemaPO.getSchemaId())) + .isEmpty(); + boolean hasModels = + !SessionUtils.getWithoutCommit( + ModelMetaMapper.class, + mapper -> mapper.listModelPOsBySchemaId(schemaPO.getSchemaId())) + .isEmpty(); + boolean hasTopics = + !SessionUtils.getWithoutCommit( + TopicMetaMapper.class, + mapper -> mapper.listTopicPOsBySchemaId(schemaPO.getSchemaId())) + .isEmpty(); + if (hasDescendantSchemas || hasTables || hasFilesets || hasModels || hasTopics) { + throw new NonEmptyEntityException( + "Entity %s has sub-entities, you should remove sub-entities first", identifier); + } + } + /** - * Collects the schema ids that participate in a cascade delete: the target schema itself plus - * every HierarchicalSchema descendant. The {@link SchemaPO} arrives in logical form (e.g. {@code - * A:B}); {@link HierarchicalConversionPOStorageOps} translates to storage form before running the - * SQL prefix match, so this method only deals in logical names. + * Collects every HierarchicalSchema descendant of the target schema. The {@link SchemaPO} arrives + * in logical form (e.g. {@code A:B}); {@link HierarchicalConversionPOStorageOps} translates to + * storage form before running the SQL prefix match. */ - private List listSchemaIdsForCascade(SchemaPO schemaPO) { + private List listDescendantSchemaPOs(SchemaPO schemaPO) { List matched = SessionUtils.getWithoutCommit( SchemaMetaMapper.class, @@ -507,16 +551,15 @@ private List listSchemaIdsForCascade(SchemaPO schemaPO) { if (matched == null || matched.isEmpty()) { return Collections.emptyList(); } - return matched.stream().map(SchemaPO::getSchemaId).collect(Collectors.toList()); + return matched.stream() + .filter(po -> !po.getSchemaId().equals(schemaPO.getSchemaId())) + .collect(Collectors.toList()); } - private void fillSchemaPOBuilderParentEntityId(SchemaPO.Builder builder, Namespace namespace) { - NamespaceUtil.checkSchema(namespace); - NamespacedEntityId namespacedEntityId = - EntityIdService.getEntityIds( - NameIdentifier.of(namespace.levels()), Entity.EntityType.CATALOG); - builder.withMetalakeId(namespacedEntityId.namespaceIds()[0]); - builder.withCatalogId(namespacedEntityId.entityId()); + private SchemaPO.Builder newSchemaPOBuilder(CatalogPO catalogPO) { + return SchemaPO.builder() + .withMetalakeId(catalogPO.getMetalakeId()) + .withCatalogId(catalogPO.getCatalogId()); } @Monitored( diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/service/UserMetaService.java b/core/src/main/java/org/apache/gravitino/storage/relational/service/UserMetaService.java index 0fbdab8ca2a..f45b77f6f0d 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/service/UserMetaService.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/service/UserMetaService.java @@ -37,13 +37,16 @@ import org.apache.gravitino.Namespace; import org.apache.gravitino.authorization.AuthorizationUtils; import org.apache.gravitino.exceptions.NoSuchEntityException; +import org.apache.gravitino.exceptions.OptimisticLockException; import org.apache.gravitino.meta.RoleEntity; import org.apache.gravitino.meta.UserEntity; import org.apache.gravitino.metrics.Monitored; +import org.apache.gravitino.storage.relational.mapper.MetalakeMetaMapper; import org.apache.gravitino.storage.relational.mapper.OwnerMetaMapper; import org.apache.gravitino.storage.relational.mapper.UserMetaMapper; import org.apache.gravitino.storage.relational.mapper.UserRoleRelMapper; import org.apache.gravitino.storage.relational.po.ExtendedUserPO; +import org.apache.gravitino.storage.relational.po.MetalakePO; import org.apache.gravitino.storage.relational.po.RolePO; import org.apache.gravitino.storage.relational.po.UserPO; import org.apache.gravitino.storage.relational.po.UserRoleRelPO; @@ -131,9 +134,18 @@ public void insertUser(UserEntity userEntity, boolean overwritten) throws IOExce try { AuthorizationUtils.checkUser(userEntity.nameIdentifier()); - Long metalakeId = - MetalakeMetaService.getInstance().getMetalakeIdByName(userEntity.namespace().level(0)); - UserPO.Builder builder = UserPO.builder().withMetalakeId(metalakeId); + String metalakeName = userEntity.namespace().level(0); + MetalakePO metalakePO = + SessionUtils.getWithoutCommit( + MetalakeMetaMapper.class, mapper -> mapper.selectMetalakeMetaByName(metalakeName)); + if (metalakePO == null) { + throw new NoSuchEntityException( + NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE, + Entity.EntityType.METALAKE.name().toLowerCase(), + metalakeName); + } + + UserPO.Builder builder = UserPO.builder().withMetalakeId(metalakePO.getMetalakeId()); UserPO userPO = POConverters.initializeUserPOWithVersion(userEntity, builder); List roleIds = Optional.ofNullable(userEntity.roleIds()).orElse(Lists.newArrayList()); @@ -141,6 +153,7 @@ public void insertUser(UserEntity userEntity, boolean overwritten) throws IOExce POConverters.initializeUserRoleRelsPOWithVersion(userEntity, roleIds); SessionUtils.doMultipleWithCommit( + () -> fenceMetalakeForUserCreate(metalakePO), () -> SessionUtils.doWithoutCommit( UserMetaMapper.class, @@ -174,12 +187,21 @@ public void insertUser(UserEntity userEntity, boolean overwritten) throws IOExce public boolean deleteUser(NameIdentifier identifier) { AuthorizationUtils.checkUser(identifier); - Long userId = EntityIdService.getEntityId(identifier, Entity.EntityType.USER); + Long metalakeId = + MetalakeMetaService.getInstance().getMetalakeIdByName(identifier.namespace().level(0)); + UserPO userPO = getUserPOByMetalakeIdAndName(metalakeId, identifier.name()); + Long userId = userPO.getUserId(); SessionUtils.doMultipleWithCommit( - () -> - SessionUtils.doWithoutCommit( - UserMetaMapper.class, mapper -> mapper.softDeleteUserMetaByUserId(userId)), + () -> { + int deleted = + SessionUtils.getWithoutCommit( + UserMetaMapper.class, + mapper -> mapper.softDeleteUserMetaByUserId(userId, userPO.getCurrentVersion())); + if (deleted == 0) { + throw optimisticLockException(identifier); + } + }, () -> SessionUtils.doWithoutCommit( UserRoleRelMapper.class, mapper -> mapper.softDeleteUserRoleRelByUserId(userId)), @@ -220,18 +242,19 @@ public UserEntity updateUser( Set insertRoleIds = Sets.difference(newRoleIds, oldRoleIds); Set deleteRoleIds = Sets.difference(oldRoleIds, newRoleIds); - if (insertRoleIds.isEmpty() && deleteRoleIds.isEmpty()) { - return newEntity; - } - try { SessionUtils.doMultipleWithCommit( - () -> - SessionUtils.doWithoutCommit( - UserMetaMapper.class, - mapper -> - mapper.updateUserMeta( - POConverters.updateUserPOWithVersion(oldUserPO, newEntity), oldUserPO)), + () -> { + int updated = + SessionUtils.getWithoutCommit( + UserMetaMapper.class, + mapper -> + mapper.updateUserMeta( + POConverters.updateUserPOWithVersion(oldUserPO, newEntity), oldUserPO)); + if (updated == 0) { + throw optimisticLockException(identifier); + } + }, () -> { if (insertRoleIds.isEmpty()) { return; @@ -370,12 +393,17 @@ public UserEntity updateUserByExternalId( try { SessionUtils.doMultipleWithCommit( - () -> - SessionUtils.doWithoutCommit( - UserMetaMapper.class, - mapper -> - mapper.updateUserMetaByExternalId( - POConverters.updateUserPOWithVersion(oldUserPO, newEntity), oldUserPO)), + () -> { + int updated = + SessionUtils.getWithoutCommit( + UserMetaMapper.class, + mapper -> + mapper.updateUserMetaByExternalId( + POConverters.updateUserPOWithVersion(oldUserPO, newEntity), oldUserPO)); + if (updated == 0) { + throw optimisticLockException(ident); + } + }, () -> SessionUtils.doWithoutCommit( UserMetaMapper.class, @@ -429,12 +457,17 @@ public UserEntity updateUserById( try { SessionUtils.doMultipleWithCommit( - () -> - SessionUtils.doWithoutCommit( - UserMetaMapper.class, - mapper -> - mapper.updateUserMeta( - POConverters.updateUserPOWithVersion(oldUserPO, newEntity), oldUserPO)), + () -> { + int updated = + SessionUtils.getWithoutCommit( + UserMetaMapper.class, + mapper -> + mapper.updateUserMeta( + POConverters.updateUserPOWithVersion(oldUserPO, newEntity), oldUserPO)); + if (updated == 0) { + throw optimisticLockException(newEntity.nameIdentifier()); + } + }, () -> SessionUtils.doWithoutCommit( UserMetaMapper.class, @@ -451,16 +484,24 @@ public UserEntity updateUserById( metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME, baseMetricName = "deleteUserById") public boolean deleteUserById(String metalake, long userId) { + UserPO userPO; try { - getUserPOByMetalakeNameAndId(metalake, userId); + userPO = getUserPOByMetalakeNameAndId(metalake, userId); } catch (NoSuchEntityException e) { return false; } + NameIdentifier identifier = AuthorizationUtils.ofUser(metalake, userPO.getUserName()); SessionUtils.doMultipleWithCommit( - () -> - SessionUtils.doWithoutCommit( - UserMetaMapper.class, mapper -> mapper.softDeleteUserMetaByUserId(userId)), + () -> { + int deleted = + SessionUtils.getWithoutCommit( + UserMetaMapper.class, + mapper -> mapper.softDeleteUserMetaByUserId(userId, userPO.getCurrentVersion())); + if (deleted == 0) { + throw optimisticLockException(identifier); + } + }, () -> SessionUtils.doWithoutCommit( UserRoleRelMapper.class, mapper -> mapper.softDeleteUserRoleRelByUserId(userId)), @@ -472,4 +513,23 @@ public boolean deleteUserById(String metalake, long userId) { userId, Entity.EntityType.USER.name()))); return true; } + + private OptimisticLockException optimisticLockException(NameIdentifier identifier) { + return new OptimisticLockException( + "The user %s was modified concurrently; retry the operation", identifier); + } + + private void fenceMetalakeForUserCreate(MetalakePO metalakePO) { + int fenced = + SessionUtils.getWithoutCommit( + MetalakeMetaMapper.class, + mapper -> + mapper.fenceMetalakeMeta( + metalakePO.getMetalakeId(), metalakePO.getCurrentVersion())); + if (fenced == 0) { + throw new OptimisticLockException( + "The parent metalake %s was modified concurrently; retry the operation", + metalakePO.getMetalakeName()); + } + } } 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 fcb05a6f466..50d1c29058f 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 @@ -136,9 +136,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()) @@ -233,9 +233,9 @@ public static CatalogPO initializeCatalogPOWithVersion( */ public static CatalogPO updateCatalogPOWithVersion( CatalogPO oldCatalogPO, CatalogEntity newCatalog, Long metalakeId) { - Long lastVersion = oldCatalogPO.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 + // catalogs do not retain independently addressable historical versions. + Long nextVersion = oldCatalogPO.getCurrentVersion() + 1; try { return CatalogPO.builder() .withCatalogId(newCatalog.id()) @@ -329,9 +329,9 @@ public static SchemaPO initializeSchemaPOWithVersion( * @return SchemaPO object with updated version */ public static SchemaPO updateSchemaPOWithVersion(SchemaPO oldSchemaPO, SchemaEntity newSchema) { - Long lastVersion = oldSchemaPO.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 + // schemas do not retain independently addressable historical versions. + Long nextVersion = oldSchemaPO.getCurrentVersion() + 1; try { return SchemaPO.builder() .withSchemaId(oldSchemaPO.getSchemaId()) @@ -977,9 +977,7 @@ public static UserPO initializeUserPOWithVersion(UserEntity userEntity, UserPO.B */ public static UserPO updateUserPOWithVersion(UserPO oldUserPO, UserEntity newUser) { Long lastVersion = oldUserPO.getLastVersion(); - // TODO: set the version to the last version + 1 when having some fields need be multiple - // version - Long nextVersion = lastVersion; + Long nextVersion = lastVersion + 1; try { return UserPO.builder() .withUserId(oldUserPO.getUserId()) @@ -1259,9 +1257,7 @@ public static GroupPO initializeGroupPOWithVersion( */ public static GroupPO updateGroupPOWithVersion(GroupPO oldGroupPO, GroupEntity newGroup) { Long lastVersion = oldGroupPO.getLastVersion(); - // TODO: set the version to the last version + 1 when having some fields need be multiple - // version - Long nextVersion = lastVersion; + Long nextVersion = lastVersion + 1; try { return GroupPO.builder() .withGroupId(oldGroupPO.getGroupId()) diff --git a/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/base/TestAuthMappers.java b/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/base/TestAuthMappers.java index e26613c7900..6698d376cb4 100644 --- a/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/base/TestAuthMappers.java +++ b/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/base/TestAuthMappers.java @@ -219,7 +219,7 @@ void testUserMetaTouchUpdatedAt() { void testUserMetaTouchUpdatedAtSkipsSoftDeleted() { insertMetalake(1L, "metalake1"); insertUser(22L, "user22", 1L); - userMetaMapper.softDeleteUserMetaByUserId(22L); + userMetaMapper.softDeleteUserMetaByUserId(22L, 1L); long beforeUpdatedAt = queryUpdatedAt("user_meta", "user_id", 22L); userMetaMapper.touchUserUpdatedAt(22L); @@ -264,7 +264,7 @@ void testGroupMetaTouchUpdatedAt() { void testGroupMetaTouchUpdatedAtSkipsSoftDeleted() { insertMetalake(1L, "metalake1"); insertGroup(31L, "group31", 1L); - groupMetaMapper.softDeleteGroupMetaByGroupId(31L); + groupMetaMapper.softDeleteGroupMetaByGroupId(31L, 1L); long beforeUpdatedAt = queryUpdatedAt("group_meta", "group_id", 31L); groupMetaMapper.touchGroupUpdatedAt(31L); @@ -287,6 +287,20 @@ void testGroupMetaGetGroupUpdatedAt() { Assertions.assertEquals(expected, info.getUpdatedAt()); } + @Test + void testPrincipalDeleteUsesCurrentVersion() { + insertMetalake(1L, "metalake1"); + insertUser(23L, "user23", 1L); + insertGroup(33L, "group33", 1L); + + Assertions.assertEquals(0, userMetaMapper.softDeleteUserMetaByUserId(23L, 2L)); + Assertions.assertEquals(0, groupMetaMapper.softDeleteGroupMetaByGroupId(33L, 2L)); + Assertions.assertNotNull(userMetaMapper.selectUserMetaByMetalakeIdAndName(1L, "user23")); + Assertions.assertNotNull(groupMetaMapper.selectGroupMetaByMetalakeIdAndName(1L, "group33")); + Assertions.assertEquals(1, userMetaMapper.softDeleteUserMetaByUserId(23L, 1L)); + Assertions.assertEquals(1, groupMetaMapper.softDeleteGroupMetaByGroupId(33L, 1L)); + } + @Test void testOwnerMetaSelectOwnerByMetadataObjectIdAndType() { insertMetalake(1L, "metalake1"); diff --git a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestCatalogMetaService.java b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestCatalogMetaService.java index de5520908af..a6468fe5f32 100644 --- a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestCatalogMetaService.java +++ b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestCatalogMetaService.java @@ -35,6 +35,8 @@ import org.apache.gravitino.EntityAlreadyExistsException; import org.apache.gravitino.NameIdentifier; import org.apache.gravitino.Namespace; +import org.apache.gravitino.exceptions.NonEmptyEntityException; +import org.apache.gravitino.exceptions.OptimisticLockException; import org.apache.gravitino.meta.AuditInfo; import org.apache.gravitino.meta.CatalogEntity; import org.apache.gravitino.meta.ColumnEntity; @@ -50,7 +52,11 @@ import org.apache.gravitino.storage.RandomIdGenerator; import org.apache.gravitino.storage.relational.TestJDBCBackend; import org.apache.gravitino.storage.relational.mapper.CatalogMetaMapper; +import org.apache.gravitino.storage.relational.mapper.MetalakeMetaMapper; +import org.apache.gravitino.storage.relational.po.CatalogPO; +import org.apache.gravitino.storage.relational.po.MetalakePO; import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper; +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; @@ -88,6 +94,40 @@ public void testInsertAlreadyExistsException() throws IOException { assertThrows(EntityAlreadyExistsException.class, () -> backend.insert(catalogCopy, false)); } + @TestTemplate + public void testInsertCatalogFencesMetalakeAndRollsBackFenceOnFailure() throws IOException { + MetalakePO beforeInsert = + SessionUtils.getWithoutCommit( + MetalakeMetaMapper.class, mapper -> mapper.selectMetalakeMetaByName(metalakeName)); + CatalogEntity catalog = + createCatalog( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofCatalog(metalakeName), + "catalog_fence", + auditInfo); + backend.insert(catalog, false); + + MetalakePO afterInsert = + SessionUtils.getWithoutCommit( + MetalakeMetaMapper.class, mapper -> mapper.selectMetalakeMetaByName(metalakeName)); + assertEquals(beforeInsert.getCurrentVersion() + 1, afterInsert.getCurrentVersion()); + assertEquals(afterInsert.getCurrentVersion(), afterInsert.getLastVersion()); + + CatalogEntity duplicate = + createCatalog( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofCatalog(metalakeName), + catalog.name(), + auditInfo); + assertThrows(EntityAlreadyExistsException.class, () -> backend.insert(duplicate, false)); + + MetalakePO afterFailure = + SessionUtils.getWithoutCommit( + MetalakeMetaMapper.class, mapper -> mapper.selectMetalakeMetaByName(metalakeName)); + assertEquals(afterInsert.getCurrentVersion(), afterFailure.getCurrentVersion()); + assertEquals(afterInsert.getLastVersion(), afterFailure.getLastVersion()); + } + @TestTemplate public void testUpdateAlreadyExistsException() throws IOException { CatalogEntity catalog = @@ -149,6 +189,121 @@ void testUpdateCatalogWithNullableComment() throws IOException { Assertions.assertNotNull(updatedCatalog.getComment()); } + @TestTemplate + public void testAlterAndDeleteUseCurrentVersion() throws IOException { + CatalogEntity catalog = + createCatalog( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofCatalog(metalakeName), + "catalog_occ", + auditInfo); + backend.insert(catalog, false); + CatalogPO oldPO = + SessionUtils.getWithoutCommit( + CatalogMetaMapper.class, mapper -> mapper.selectCatalogMetaById(catalog.id())); + CatalogEntity updatedCatalog = + CatalogEntity.builder() + .withId(catalog.id()) + .withName(catalog.name()) + .withNamespace(catalog.namespace()) + .withAuditInfo(auditInfo) + .withComment("updated") + .withProperties(catalog.getProperties()) + .withType(catalog.getType()) + .withProvider(catalog.getProvider()) + .build(); + CatalogPO newPO = + POConverters.updateCatalogPOWithVersion(oldPO, updatedCatalog, oldPO.getMetalakeId()); + + int updated = + SessionUtils.doWithCommitAndFetchResult( + CatalogMetaMapper.class, mapper -> mapper.updateCatalogMeta(newPO, oldPO)); + int staleUpdate = + SessionUtils.doWithCommitAndFetchResult( + CatalogMetaMapper.class, mapper -> mapper.updateCatalogMeta(newPO, oldPO)); + int staleDelete = + SessionUtils.doWithCommitAndFetchResult( + CatalogMetaMapper.class, + mapper -> + mapper.softDeleteCatalogMetasByCatalogId(catalog.id(), oldPO.getCurrentVersion())); + assertEquals(1, updated); + assertEquals(0, staleUpdate); + assertEquals(0, staleDelete); + assertTrue(backend.exists(catalog.nameIdentifier(), Entity.EntityType.CATALOG)); + int deleted = + SessionUtils.doWithCommitAndFetchResult( + CatalogMetaMapper.class, + mapper -> + mapper.softDeleteCatalogMetasByCatalogId(catalog.id(), newPO.getCurrentVersion())); + assertEquals(1, deleted); + } + + @TestTemplate + public void testAlterReportsOptimisticLockConflict() throws IOException { + CatalogEntity catalog = + createCatalog( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofCatalog(metalakeName), + "catalog_alter_conflict", + auditInfo); + backend.insert(catalog, false); + + assertThrows( + OptimisticLockException.class, + () -> + CatalogMetaService.getInstance() + .updateCatalog( + catalog.nameIdentifier(), + entity -> { + CatalogEntity current = (CatalogEntity) entity; + CatalogPO currentPO = + SessionUtils.getWithoutCommit( + CatalogMetaMapper.class, + mapper -> mapper.selectCatalogMetaById(current.id())); + CatalogEntity competingUpdate = + copyCatalogWithComment(current, "competing update"); + CatalogPO competingPO = + POConverters.updateCatalogPOWithVersion( + currentPO, competingUpdate, currentPO.getMetalakeId()); + SessionUtils.doWithCommitAndFetchResult( + CatalogMetaMapper.class, + mapper -> mapper.updateCatalogMeta(competingPO, currentPO)); + return copyCatalogWithComment(current, "requested update"); + })); + } + + @TestTemplate + public void testNonCascadeDeleteRollsBackCatalogFence() throws IOException { + CatalogEntity catalog = + createCatalog( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofCatalog(metalakeName), + "catalog_non_empty", + auditInfo); + backend.insert(catalog, false); + SchemaEntity schema = + createSchemaEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofSchema(metalakeName, catalog.name()), + "schema", + auditInfo); + backend.insert(schema, false); + CatalogPO beforeDelete = + SessionUtils.getWithoutCommit( + CatalogMetaMapper.class, mapper -> mapper.selectCatalogMetaById(catalog.id())); + + assertThrows( + NonEmptyEntityException.class, + () -> CatalogMetaService.getInstance().deleteCatalog(catalog.nameIdentifier(), false)); + + CatalogPO afterDelete = + SessionUtils.getWithoutCommit( + CatalogMetaMapper.class, mapper -> mapper.selectCatalogMetaById(catalog.id())); + assertEquals(beforeDelete.getCurrentVersion(), afterDelete.getCurrentVersion()); + assertTrue(backend.exists(catalog.nameIdentifier(), Entity.EntityType.CATALOG)); + assertTrue(backend.exists(schema.nameIdentifier(), Entity.EntityType.SCHEMA)); + } + @TestTemplate public void testMetaLifeCycleFromCreationToDeletion() throws IOException { CatalogEntity catalog = @@ -303,6 +458,19 @@ public void testDeleteCatalogCascadeRemovesTagRelations() throws IOException { assertEquals(0, countActiveTagRelForMetadataObject(function.id(), "FUNCTION")); } + private CatalogEntity copyCatalogWithComment(CatalogEntity catalog, String comment) { + return CatalogEntity.builder() + .withId(catalog.id()) + .withName(catalog.name()) + .withNamespace(catalog.namespace()) + .withType(catalog.getType()) + .withProvider(catalog.getProvider()) + .withComment(comment) + .withProperties(catalog.getProperties()) + .withAuditInfo(auditInfo) + .build(); + } + private void associateTag(TagEntity tag, NameIdentifier ident, Entity.EntityType type) throws IOException { TagMetaService.getInstance() diff --git a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestGroupMetaService.java b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestGroupMetaService.java index 801e8dadb5f..3ebd2cf8d5d 100644 --- a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestGroupMetaService.java +++ b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestGroupMetaService.java @@ -42,6 +42,7 @@ import org.apache.gravitino.Namespace; import org.apache.gravitino.authorization.AuthorizationUtils; import org.apache.gravitino.exceptions.NoSuchEntityException; +import org.apache.gravitino.exceptions.OptimisticLockException; import org.apache.gravitino.meta.AuditInfo; import org.apache.gravitino.meta.BaseMetalake; import org.apache.gravitino.meta.GroupEntity; @@ -49,7 +50,10 @@ import org.apache.gravitino.storage.RandomIdGenerator; import org.apache.gravitino.storage.relational.TestJDBCBackend; import org.apache.gravitino.storage.relational.mapper.GroupMetaMapper; +import org.apache.gravitino.storage.relational.mapper.MetalakeMetaMapper; import org.apache.gravitino.storage.relational.mapper.RoleMetaMapper; +import org.apache.gravitino.storage.relational.po.GroupPO; +import org.apache.gravitino.storage.relational.po.MetalakePO; import org.apache.gravitino.storage.relational.po.RolePO; import org.apache.gravitino.storage.relational.po.auth.GroupUpdatedAt; import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper; @@ -707,8 +711,8 @@ void testUpdateGroup() throws IOException { Assertions.assertEquals("creator", grantRevokeGroup.auditInfo().creator()); Assertions.assertEquals("grantRevokeUser", grantRevokeGroup.auditInfo().lastModifier()); - // no update - Function noUpdater = + // metadata-only update + Function metadataUpdater = group -> { AuditInfo updateAuditInfo = AuditInfo.builder() @@ -730,19 +734,20 @@ void testUpdateGroup() throws IOException { .withAuditInfo(updateAuditInfo) .build(); }; - long beforeNoUpdate = getGroupUpdatedAt(group1.name()).getUpdatedAt(); - Assertions.assertNotNull(groupMetaService.updateGroup(group1.nameIdentifier(), noUpdater)); - Assertions.assertEquals(beforeNoUpdate, getGroupUpdatedAt(group1.name()).getUpdatedAt()); - GroupEntity noUpdaterGroup = + long beforeMetadataUpdate = getGroupUpdatedAt(group1.name()).getUpdatedAt(); + Assertions.assertNotNull( + groupMetaService.updateGroup(group1.nameIdentifier(), metadataUpdater)); + Assertions.assertTrue(getGroupUpdatedAt(group1.name()).getUpdatedAt() >= beforeMetadataUpdate); + GroupEntity metadataUpdatedGroup = GroupMetaService.getInstance().getGroupByIdentifier(group1.nameIdentifier()); - Assertions.assertEquals(group1.id(), noUpdaterGroup.id()); - Assertions.assertEquals(group1.name(), noUpdaterGroup.name()); + Assertions.assertEquals(group1.id(), metadataUpdatedGroup.id()); + Assertions.assertEquals(group1.name(), metadataUpdatedGroup.name()); Assertions.assertEquals( - Sets.newHashSet("role1", "role4"), Sets.newHashSet(noUpdaterGroup.roleNames())); + Sets.newHashSet("role1", "role4"), Sets.newHashSet(metadataUpdatedGroup.roleNames())); Assertions.assertEquals( - Sets.newHashSet(role1.id(), role4.id()), Sets.newHashSet(noUpdaterGroup.roleIds())); - Assertions.assertEquals("creator", noUpdaterGroup.auditInfo().creator()); - Assertions.assertEquals("grantRevokeUser", noUpdaterGroup.auditInfo().lastModifier()); + Sets.newHashSet(role1.id(), role4.id()), Sets.newHashSet(metadataUpdatedGroup.roleIds())); + Assertions.assertEquals("creator", metadataUpdatedGroup.auditInfo().creator()); + Assertions.assertEquals("noUpdateUser", metadataUpdatedGroup.auditInfo().lastModifier()); // Delete a role, the group entity won't contain this role. RoleMetaService.getInstance().deleteRole(role1.nameIdentifier()); @@ -1079,6 +1084,64 @@ void testGroupExtId() throws IOException { IllegalArgumentException.class, () -> svc.getGroupByExternalId(groupExtIdent(""))); } + @TestTemplate + void testConcurrentUpdateDoesNotChangeRolesOnConflict() throws IOException { + createAndInsertMakeLake(metalakeName); + createAndInsertCatalog(metalakeName, catalogName); + RoleEntity role1 = + createRoleEntity( + RandomIdGenerator.INSTANCE.nextId(), + AuthorizationUtils.ofRoleNamespace(metalakeName), + "role1", + AUDIT_INFO, + catalogName); + RoleEntity role2 = + createRoleEntity( + RandomIdGenerator.INSTANCE.nextId(), + AuthorizationUtils.ofRoleNamespace(metalakeName), + "role2", + AUDIT_INFO, + catalogName); + RoleMetaService.getInstance().insertRole(role1, false); + RoleMetaService.getInstance().insertRole(role2, false); + GroupEntity group = + createGroupEntity( + RandomIdGenerator.INSTANCE.nextId(), + AuthorizationUtils.ofGroupNamespace(metalakeName), + "concurrent-group", + AUDIT_INFO, + Lists.newArrayList(role1.name()), + Lists.newArrayList(role1.id())); + GroupMetaService.getInstance().insertGroup(group, false); + + Assertions.assertThrows( + OptimisticLockException.class, + () -> + GroupMetaService.getInstance() + .updateGroup( + group.nameIdentifier(), + (GroupEntity oldGroup) -> { + advanceGroupVersion(group.id()); + List roleNames = Lists.newArrayList(oldGroup.roleNames()); + List roleIds = Lists.newArrayList(oldGroup.roleIds()); + roleNames.add(role2.name()); + roleIds.add(role2.id()); + return GroupEntity.builder() + .withId(oldGroup.id()) + .withName(oldGroup.name()) + .withNamespace(oldGroup.namespace()) + .withExternalId(oldGroup.externalId()) + .withRoleNames(roleNames) + .withRoleIds(roleIds) + .withAuditInfo(oldGroup.auditInfo()) + .build(); + })); + + GroupEntity storedGroup = + GroupMetaService.getInstance().getGroupByIdentifier(group.nameIdentifier()); + assertEquals(Sets.newHashSet(role1.id()), Sets.newHashSet(storedGroup.roleIds())); + } + @TestTemplate void testExtDup() throws IOException { GroupMetaService svc = groupMetaService(); @@ -1088,6 +1151,86 @@ void testExtDup() throws IOException { () -> svc.insertGroup(groupWithExtId("g2", "ext-1"), false)); } + @TestTemplate + void testCreateFencesMetalakeAndRollsBackFenceOnFailure() throws IOException { + createAndInsertMakeLake(metalakeName); + GroupMetaService service = GroupMetaService.getInstance(); + MetalakePO beforeCreate = getMetalakePO(); + GroupEntity group = groupWithExtId("fenced-group", "fenced-group-ext-id"); + + service.insertGroup(group, false); + + MetalakePO afterCreate = getMetalakePO(); + assertEquals(beforeCreate.getCurrentVersion() + 1, afterCreate.getCurrentVersion()); + assertEquals(afterCreate.getCurrentVersion(), afterCreate.getLastVersion()); + + GroupEntity duplicate = groupWithExtId(group.name(), "another-ext-id"); + Assertions.assertThrows( + EntityAlreadyExistsException.class, () -> service.insertGroup(duplicate, false)); + + MetalakePO afterFailedCreate = getMetalakePO(); + assertEquals(afterCreate.getCurrentVersion(), afterFailedCreate.getCurrentVersion()); + assertEquals(afterCreate.getLastVersion(), afterFailedCreate.getLastVersion()); + } + + @TestTemplate + void testMetadataOnlyUpdateUsesOcc() throws IOException { + GroupMetaService service = groupMetaService(); + GroupEntity group = groupWithExtId("metadata-only-group", "metadata-only-ext-id"); + service.insertGroup(group, false); + GroupPO beforeUpdate = getGroupPO(group.name()); + + service.updateGroup( + group.nameIdentifier(), (GroupEntity oldGroup) -> copyGroup(oldGroup, "updated-ext-id")); + + GroupPO afterUpdate = getGroupPO(group.name()); + assertEquals(beforeUpdate.getCurrentVersion() + 1, afterUpdate.getCurrentVersion()); + assertEquals( + "updated-ext-id", service.getGroupByIdentifier(group.nameIdentifier()).externalId()); + + Assertions.assertThrows( + OptimisticLockException.class, + () -> + service.updateGroup( + group.nameIdentifier(), + (GroupEntity oldGroup) -> { + advanceGroupVersion(group.id()); + return copyGroup(oldGroup, "conflicting-ext-id"); + })); + assertEquals( + "updated-ext-id", service.getGroupByIdentifier(group.nameIdentifier()).externalId()); + } + + @TestTemplate + void testByIdMutationsUseOcc() throws IOException { + GroupMetaService service = groupMetaService(); + GroupEntity group = groupWithExtId("by-id-group", "by-id-ext-id"); + service.insertGroup(group, false); + GroupPO beforeUpdate = getGroupPO(group.name()); + + service.updateGroupById( + metalakeName, group.id(), (GroupEntity oldGroup) -> copyGroup(oldGroup, "updated-by-id")); + + GroupPO afterUpdate = getGroupPO(group.name()); + assertEquals(beforeUpdate.getCurrentVersion() + 1, afterUpdate.getCurrentVersion()); + assertEquals("updated-by-id", service.getGroupById(metalakeName, group.id()).externalId()); + + Assertions.assertThrows( + OptimisticLockException.class, + () -> + service.updateGroupById( + metalakeName, + group.id(), + (GroupEntity oldGroup) -> { + advanceGroupVersion(group.id()); + return copyGroup(oldGroup, "conflicting-by-id"); + })); + assertEquals("updated-by-id", service.getGroupById(metalakeName, group.id()).externalId()); + + assertTrue(service.deleteGroupById(metalakeName, group.id())); + assertFalse(service.deleteGroupById(metalakeName, group.id())); + } + @TestTemplate void testGroupExtDel() throws IOException { GroupMetaService svc = groupMetaService(); @@ -1109,6 +1252,30 @@ private GroupMetaService groupMetaService() throws IOException { return GroupMetaService.getInstance(); } + private MetalakePO getMetalakePO() { + return SessionUtils.getWithoutCommit( + MetalakeMetaMapper.class, mapper -> mapper.selectMetalakeMetaByName(metalakeName)); + } + + private GroupPO getGroupPO(String groupName) { + MetalakePO metalakePO = getMetalakePO(); + return SessionUtils.getWithoutCommit( + GroupMetaMapper.class, + mapper -> mapper.selectGroupMetaByMetalakeIdAndName(metalakePO.getMetalakeId(), groupName)); + } + + private GroupEntity copyGroup(GroupEntity group, String externalId) { + return GroupEntity.builder() + .withId(group.id()) + .withName(group.name()) + .withNamespace(group.namespace()) + .withExternalId(externalId) + .withRoleNames(group.roleNames()) + .withRoleIds(group.roleIds()) + .withAuditInfo(group.auditInfo()) + .build(); + } + private void assertThrowsExt(Class type, Executable executable) { Assertions.assertThrows(type, executable); } @@ -1134,4 +1301,19 @@ private GroupEntity createGroupEntity( .withAuditInfo(auditInfo) .build(); } + + private void advanceGroupVersion(long groupId) { + try (SqlSession sqlSession = + SqlSessionFactoryHelper.getInstance().getSqlSessionFactory().openSession(true); + Connection connection = sqlSession.getConnection(); + Statement statement = connection.createStatement()) { + assertEquals( + 1, + statement.executeUpdate( + "UPDATE group_meta SET current_version = current_version + 1 WHERE group_id = " + + groupId)); + } catch (SQLException e) { + throw new RuntimeException("Advance group version failed", e); + } + } } 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..d72650e8854 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,16 @@ import java.util.List; import org.apache.gravitino.Entity; import org.apache.gravitino.EntityAlreadyExistsException; +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 +98,140 @@ 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 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/service/TestSchemaMetaService.java b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestSchemaMetaService.java index a1e43144ec8..9e8a84ca566 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 @@ -37,6 +37,8 @@ import org.apache.gravitino.NameIdentifier; import org.apache.gravitino.Namespace; import org.apache.gravitino.exceptions.NonEmptyEntityException; +import org.apache.gravitino.exceptions.OptimisticLockException; +import org.apache.gravitino.meta.CatalogEntity; import org.apache.gravitino.meta.ColumnEntity; import org.apache.gravitino.meta.FilesetEntity; import org.apache.gravitino.meta.FunctionEntity; @@ -49,7 +51,13 @@ import org.apache.gravitino.rel.types.Types; import org.apache.gravitino.storage.RandomIdGenerator; import org.apache.gravitino.storage.relational.TestJDBCBackend; +import org.apache.gravitino.storage.relational.mapper.CatalogMetaMapper; +import org.apache.gravitino.storage.relational.mapper.SchemaMetaMapper; +import org.apache.gravitino.storage.relational.po.CatalogPO; +import org.apache.gravitino.storage.relational.po.SchemaPO; import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper; +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; import org.apache.ibatis.session.SqlSession; @@ -81,6 +89,42 @@ public void testInsertAlreadyExistsException() throws IOException { assertThrows(EntityAlreadyExistsException.class, () -> backend.insert(schemaCopy, false)); } + @TestTemplate + public void testInsertSchemaFencesCatalogAndRollsBackFenceOnFailure() throws IOException { + createAndInsertMakeLake(metalakeName); + CatalogEntity catalog = createAndInsertCatalog(metalakeName, catalogName); + CatalogPO beforeInsert = + SessionUtils.getWithoutCommit( + CatalogMetaMapper.class, mapper -> mapper.selectCatalogMetaById(catalog.id())); + SchemaEntity schema = + createSchemaEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofSchema(metalakeName, catalogName), + "schema_fence", + AUDIT_INFO); + backend.insert(schema, false); + + CatalogPO afterInsert = + SessionUtils.getWithoutCommit( + CatalogMetaMapper.class, mapper -> mapper.selectCatalogMetaById(catalog.id())); + Assertions.assertEquals(beforeInsert.getCurrentVersion() + 1, afterInsert.getCurrentVersion()); + Assertions.assertEquals(afterInsert.getCurrentVersion(), afterInsert.getLastVersion()); + + SchemaEntity duplicate = + createSchemaEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofSchema(metalakeName, catalogName), + schema.name(), + AUDIT_INFO); + assertThrows(EntityAlreadyExistsException.class, () -> backend.insert(duplicate, false)); + + CatalogPO afterFailure = + SessionUtils.getWithoutCommit( + CatalogMetaMapper.class, mapper -> mapper.selectCatalogMetaById(catalog.id())); + Assertions.assertEquals(afterInsert.getCurrentVersion(), afterFailure.getCurrentVersion()); + Assertions.assertEquals(afterInsert.getLastVersion(), afterFailure.getLastVersion()); + } + @TestTemplate public void testUpdateAlreadyExistsException() throws IOException { createAndInsertMakeLake(metalakeName); @@ -145,6 +189,91 @@ public void testUpdateSchemaCommentFromNull() throws IOException { Assertions.assertEquals("schema comment updated", updatedSchema.comment()); } + @TestTemplate + public void testAlterAndDeleteUseCurrentVersion() throws IOException { + createAndInsertMakeLake(metalakeName); + createAndInsertCatalog(metalakeName, catalogName); + SchemaEntity schema = + createSchemaEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofSchema(metalakeName, catalogName), + "schema_occ", + AUDIT_INFO); + backend.insert(schema, false); + SchemaPO oldPO = + SessionUtils.getWithoutCommit( + SchemaMetaMapper.class, mapper -> mapper.selectSchemaMetaById(schema.id())); + SchemaEntity updatedSchema = + SchemaEntity.builder() + .withId(schema.id()) + .withName(schema.name()) + .withNamespace(schema.namespace()) + .withAuditInfo(schema.auditInfo()) + .withComment("updated") + .withProperties(schema.properties()) + .build(); + SchemaPO newPO = POConverters.updateSchemaPOWithVersion(oldPO, updatedSchema); + + int updated = + SessionUtils.doWithCommitAndFetchResult( + SchemaMetaMapper.class, mapper -> mapper.updateSchemaMeta(newPO, oldPO)); + int staleUpdate = + SessionUtils.doWithCommitAndFetchResult( + SchemaMetaMapper.class, mapper -> mapper.updateSchemaMeta(newPO, oldPO)); + int staleDelete = + SessionUtils.doWithCommitAndFetchResult( + SchemaMetaMapper.class, + mapper -> + mapper.softDeleteSchemaMetaBySchemaIdAndVersion( + schema.id(), oldPO.getCurrentVersion())); + Assertions.assertEquals(1, updated); + Assertions.assertEquals(0, staleUpdate); + Assertions.assertEquals(0, staleDelete); + assertTrue(backend.exists(schema.nameIdentifier(), Entity.EntityType.SCHEMA)); + int deleted = + SessionUtils.doWithCommitAndFetchResult( + SchemaMetaMapper.class, + mapper -> + mapper.softDeleteSchemaMetaBySchemaIdAndVersion( + schema.id(), newPO.getCurrentVersion())); + Assertions.assertEquals(1, deleted); + } + + @TestTemplate + public void testAlterReportsOptimisticLockConflict() throws IOException { + createAndInsertMakeLake(metalakeName); + createAndInsertCatalog(metalakeName, catalogName); + SchemaEntity schema = + createSchemaEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofSchema(metalakeName, catalogName), + "schema_alter_conflict", + AUDIT_INFO); + backend.insert(schema, false); + + assertThrows( + OptimisticLockException.class, + () -> + SchemaMetaService.getInstance() + .updateSchema( + schema.nameIdentifier(), + entity -> { + SchemaEntity current = (SchemaEntity) entity; + SchemaPO currentPO = + SessionUtils.getWithoutCommit( + SchemaMetaMapper.class, + mapper -> mapper.selectSchemaMetaById(current.id())); + SchemaEntity competingUpdate = + copySchemaWithComment(current, "competing update"); + SchemaPO competingPO = + POConverters.updateSchemaPOWithVersion(currentPO, competingUpdate); + SessionUtils.doWithCommitAndFetchResult( + SchemaMetaMapper.class, + mapper -> mapper.updateSchemaMeta(competingPO, currentPO)); + return copySchemaWithComment(current, "requested update"); + })); + } + @TestTemplate public void testMetaLifeCycleFromCreationToDeletion() throws IOException { createAndInsertMakeLake(metalakeName); @@ -215,12 +344,22 @@ public void testDeleteSchemaNonCascadingFailsWhenTopicExists() throws IOExceptio topicName, AUDIT_INFO); topicMetaService.insertTopic(topic, false); + SchemaPO beforeDelete = + SessionUtils.getWithoutCommit( + SchemaMetaMapper.class, mapper -> mapper.selectSchemaMetaById(schema.id())); Assertions.assertThrows( NonEmptyEntityException.class, () -> schemaMetaService.deleteSchema(schema.nameIdentifier(), false), "Non-cascading delete must fail when dependent topics exist."); + SchemaPO afterDelete = + SessionUtils.getWithoutCommit( + SchemaMetaMapper.class, mapper -> mapper.selectSchemaMetaById(schema.id())); + Assertions.assertEquals(beforeDelete.getCurrentVersion(), afterDelete.getCurrentVersion()); + assertTrue(backend.exists(schema.nameIdentifier(), Entity.EntityType.SCHEMA)); + assertTrue(backend.exists(topic.nameIdentifier(), Entity.EntityType.TOPIC)); + topicMetaService.deleteTopic(topic.nameIdentifier()); schemaMetaService.deleteSchema(schema.nameIdentifier(), false); } @@ -507,14 +646,24 @@ public void testInsertHierarchicalSecondLeafReusesAncestorsWithoutUpsert() throw .build(); schemaMetaService.insertSchema(first, false); - long idA = - schemaMetaService - .getSchemaByIdentifier(NameIdentifier.of(metalakeName, catalogName, ancestorA)) - .id(); - long idAB = - schemaMetaService - .getSchemaByIdentifier(NameIdentifier.of(metalakeName, catalogName, ancestorAB)) - .id(); + SchemaPO ancestorAPOBefore = + SessionUtils.getWithoutCommit( + SchemaMetaMapper.class, + mapper -> + mapper.selectSchemaMetaById( + schemaMetaService + .getSchemaByIdentifier( + NameIdentifier.of(metalakeName, catalogName, ancestorA)) + .id())); + SchemaPO ancestorABPOBefore = + SessionUtils.getWithoutCommit( + SchemaMetaMapper.class, + mapper -> + mapper.selectSchemaMetaById( + schemaMetaService + .getSchemaByIdentifier( + NameIdentifier.of(metalakeName, catalogName, ancestorAB)) + .id())); SchemaEntity second = SchemaEntity.builder() @@ -527,16 +676,41 @@ public void testInsertHierarchicalSecondLeafReusesAncestorsWithoutUpsert() throw .build(); schemaMetaService.insertSchema(second, false); + SchemaPO ancestorAPOAfter = + SessionUtils.getWithoutCommit( + SchemaMetaMapper.class, + mapper -> + mapper.selectSchemaMetaById( + schemaMetaService + .getSchemaByIdentifier( + NameIdentifier.of(metalakeName, catalogName, ancestorA)) + .id())); + SchemaPO ancestorABPOAfter = + SessionUtils.getWithoutCommit( + SchemaMetaMapper.class, + mapper -> + mapper.selectSchemaMetaById( + schemaMetaService + .getSchemaByIdentifier( + NameIdentifier.of(metalakeName, catalogName, ancestorAB)) + .id())); + Assertions.assertEquals(ancestorAPOBefore.getSchemaId(), ancestorAPOAfter.getSchemaId()); + Assertions.assertEquals(ancestorABPOBefore.getSchemaId(), ancestorABPOAfter.getSchemaId()); Assertions.assertEquals( - idA, - schemaMetaService - .getSchemaByIdentifier(NameIdentifier.of(metalakeName, catalogName, ancestorA)) - .id()); + ancestorAPOBefore.getCurrentVersion() + 1, ancestorAPOAfter.getCurrentVersion()); Assertions.assertEquals( - idAB, - schemaMetaService - .getSchemaByIdentifier(NameIdentifier.of(metalakeName, catalogName, ancestorAB)) - .id()); + ancestorABPOBefore.getCurrentVersion() + 1, ancestorABPOAfter.getCurrentVersion()); + } + + private SchemaEntity copySchemaWithComment(SchemaEntity schema, String comment) { + return SchemaEntity.builder() + .withId(schema.id()) + .withName(schema.name()) + .withNamespace(schema.namespace()) + .withComment(comment) + .withProperties(schema.properties()) + .withAuditInfo(schema.auditInfo()) + .build(); } private void associateTag(TagEntity tag, NameIdentifier ident, Entity.EntityType type) diff --git a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestUserMetaService.java b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestUserMetaService.java index f6370317a9d..9e282ca6fba 100644 --- a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestUserMetaService.java +++ b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestUserMetaService.java @@ -44,6 +44,7 @@ import org.apache.gravitino.Namespace; import org.apache.gravitino.authorization.AuthorizationUtils; import org.apache.gravitino.exceptions.NoSuchEntityException; +import org.apache.gravitino.exceptions.OptimisticLockException; import org.apache.gravitino.meta.AuditInfo; import org.apache.gravitino.meta.BaseMetalake; import org.apache.gravitino.meta.CatalogEntity; @@ -56,9 +57,12 @@ import org.apache.gravitino.meta.UserEntity; 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.mapper.RoleMetaMapper; import org.apache.gravitino.storage.relational.mapper.UserMetaMapper; +import org.apache.gravitino.storage.relational.po.MetalakePO; import org.apache.gravitino.storage.relational.po.RolePO; +import org.apache.gravitino.storage.relational.po.UserPO; import org.apache.gravitino.storage.relational.po.auth.AuthPrefetchRow; import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper; import org.apache.gravitino.storage.relational.utils.SessionUtils; @@ -832,7 +836,7 @@ void updateUser() throws IOException { Assertions.assertEquals("creator", grantRevokeUser.auditInfo().creator()); Assertions.assertEquals("grantRevokeUser", grantRevokeUser.auditInfo().lastModifier()); - Function noUpdater = + Function metadataUpdater = user -> { AuditInfo updateAuditInfo = AuditInfo.builder() @@ -854,17 +858,17 @@ void updateUser() throws IOException { .withAuditInfo(updateAuditInfo) .build(); }; - Assertions.assertNotNull(userMetaService.updateUser(user1.nameIdentifier(), noUpdater)); - UserEntity noUpdaterUser = + Assertions.assertNotNull(userMetaService.updateUser(user1.nameIdentifier(), metadataUpdater)); + UserEntity metadataUpdatedUser = UserMetaService.getInstance().getUserByIdentifier(user1.nameIdentifier()); - Assertions.assertEquals(user1.id(), noUpdaterUser.id()); - Assertions.assertEquals(user1.name(), noUpdaterUser.name()); + Assertions.assertEquals(user1.id(), metadataUpdatedUser.id()); + Assertions.assertEquals(user1.name(), metadataUpdatedUser.name()); Assertions.assertEquals( - Sets.newHashSet("role1", "role4"), Sets.newHashSet(noUpdaterUser.roleNames())); + Sets.newHashSet("role1", "role4"), Sets.newHashSet(metadataUpdatedUser.roleNames())); Assertions.assertEquals( - Sets.newHashSet(role1.id(), role4.id()), Sets.newHashSet(noUpdaterUser.roleIds())); - Assertions.assertEquals("creator", noUpdaterUser.auditInfo().creator()); - Assertions.assertEquals("grantRevokeUser", noUpdaterUser.auditInfo().lastModifier()); + Sets.newHashSet(role1.id(), role4.id()), Sets.newHashSet(metadataUpdatedUser.roleIds())); + Assertions.assertEquals("creator", metadataUpdatedUser.auditInfo().creator()); + Assertions.assertEquals("noUpdateUser", metadataUpdatedUser.auditInfo().lastModifier()); // Delete a role, the user entity won't contain this role. RoleMetaService.getInstance().deleteRole(role1.nameIdentifier()); @@ -1331,11 +1335,196 @@ void testExtDup() throws IOException { () -> svc.insertUser(userWithExtId("u2", "ext-1"), false)); } + @TestTemplate + void testCreateFencesMetalakeAndRollsBackFenceOnFailure() throws IOException { + createAndInsertMakeLake(metalakeName); + UserMetaService service = UserMetaService.getInstance(); + MetalakePO beforeCreate = getMetalakePO(); + UserEntity user = userWithExtId("fenced-user", "fenced-user-ext-id"); + + service.insertUser(user, false); + + MetalakePO afterCreate = getMetalakePO(); + assertEquals(beforeCreate.getCurrentVersion() + 1, afterCreate.getCurrentVersion()); + assertEquals(afterCreate.getCurrentVersion(), afterCreate.getLastVersion()); + + UserEntity duplicate = userWithExtId(user.name(), "another-ext-id"); + Assertions.assertThrows( + EntityAlreadyExistsException.class, () -> service.insertUser(duplicate, false)); + + MetalakePO afterFailedCreate = getMetalakePO(); + assertEquals(afterCreate.getCurrentVersion(), afterFailedCreate.getCurrentVersion()); + assertEquals(afterCreate.getLastVersion(), afterFailedCreate.getLastVersion()); + } + + @TestTemplate + void testMetadataOnlyUpdateUsesOcc() throws IOException { + UserMetaService service = userMetaService(); + UserEntity user = userWithExtId("metadata-only-user", "metadata-only-ext-id"); + service.insertUser(user, false); + UserPO beforeUpdate = getUserPO(user.name()); + + service.updateUser(user.nameIdentifier(), enabledUpdater(false)); + + UserPO afterUpdate = getUserPO(user.name()); + assertEquals(beforeUpdate.getCurrentVersion() + 1, afterUpdate.getCurrentVersion()); + assertFalse(service.getUserByIdentifier(user.nameIdentifier()).enabled()); + + Assertions.assertThrows( + OptimisticLockException.class, + () -> + service.updateUser( + user.nameIdentifier(), + (UserEntity oldUser) -> { + advanceUserVersion(user.id()); + return enabledUpdater(true).apply(oldUser); + })); + assertFalse(service.getUserByIdentifier(user.nameIdentifier()).enabled()); + } + + @TestTemplate + void testConcurrentUpdateDoesNotChangeRolesOnConflict() throws IOException { + createAndInsertMakeLake(metalakeName); + createAndInsertCatalog(metalakeName, "catalog"); + RoleEntity role1 = + createRoleEntity( + RandomIdGenerator.INSTANCE.nextId(), + AuthorizationUtils.ofRoleNamespace(metalakeName), + "role1", + AUDIT_INFO, + "catalog"); + RoleEntity role2 = + createRoleEntity( + RandomIdGenerator.INSTANCE.nextId(), + AuthorizationUtils.ofRoleNamespace(metalakeName), + "role2", + AUDIT_INFO, + "catalog"); + RoleMetaService.getInstance().insertRole(role1, false); + RoleMetaService.getInstance().insertRole(role2, false); + UserEntity user = + createUserEntity( + RandomIdGenerator.INSTANCE.nextId(), + AuthorizationUtils.ofUserNamespace(metalakeName), + "concurrent-user", + AUDIT_INFO, + Lists.newArrayList(role1.name()), + Lists.newArrayList(role1.id())); + UserMetaService.getInstance().insertUser(user, false); + + Assertions.assertThrows( + OptimisticLockException.class, + () -> + UserMetaService.getInstance() + .updateUser( + user.nameIdentifier(), + (UserEntity oldUser) -> { + advanceUserVersion(user.id()); + List roleNames = Lists.newArrayList(oldUser.roleNames()); + List roleIds = Lists.newArrayList(oldUser.roleIds()); + roleNames.add(role2.name()); + roleIds.add(role2.id()); + return UserEntity.builder() + .withId(oldUser.id()) + .withName(oldUser.name()) + .withNamespace(oldUser.namespace()) + .withExternalId(oldUser.externalId()) + .withEnabled(oldUser.enabled()) + .withRoleNames(roleNames) + .withRoleIds(roleIds) + .withAuditInfo(oldUser.auditInfo()) + .build(); + })); + + UserEntity storedUser = + UserMetaService.getInstance().getUserByIdentifier(user.nameIdentifier()); + assertEquals(Sets.newHashSet(role1.id()), Sets.newHashSet(storedUser.roleIds())); + } + + @TestTemplate + void testConcurrentExternalIdUpdateRollsBackOnConflict() throws IOException { + UserMetaService service = userMetaService(); + UserEntity user = userWithExtId("concurrent-user", "concurrent-ext-id"); + service.insertUser(user, false); + + Assertions.assertThrows( + OptimisticLockException.class, + () -> + service.updateUserByExternalId( + userExtIdent(user.externalId()), + (UserEntity oldUser) -> { + advanceUserVersion(user.id()); + return UserEntity.builder() + .withId(oldUser.id()) + .withName(oldUser.name()) + .withNamespace(oldUser.namespace()) + .withExternalId(oldUser.externalId()) + .withEnabled(false) + .withRoleNames(oldUser.roleNames()) + .withRoleIds(oldUser.roleIds()) + .withAuditInfo(oldUser.auditInfo()) + .build(); + })); + + assertTrue(queryEnabledByExtId(user.externalId())); + } + + @TestTemplate + void testConcurrentByIdUpdateRollsBackOnConflict() throws IOException { + UserMetaService service = userMetaService(); + UserEntity user = userWithExtId("concurrent-by-id-user", "concurrent-by-id-ext-id"); + service.insertUser(user, false); + + Assertions.assertThrows( + OptimisticLockException.class, + () -> + service.updateUserById( + metalakeName, + user.id(), + (UserEntity oldUser) -> { + advanceUserVersion(user.id()); + return UserEntity.builder() + .withId(oldUser.id()) + .withName(oldUser.name()) + .withNamespace(oldUser.namespace()) + .withExternalId(oldUser.externalId()) + .withEnabled(false) + .withRoleNames(oldUser.roleNames()) + .withRoleIds(oldUser.roleIds()) + .withAuditInfo(oldUser.auditInfo()) + .build(); + })); + + assertTrue(queryEnabledByExtId(user.externalId())); + } + + @TestTemplate + void testDeleteUserById() throws IOException { + UserMetaService service = userMetaService(); + UserEntity user = userWithExtId("delete-by-id-user", "delete-by-id-ext-id"); + service.insertUser(user, false); + + assertTrue(service.deleteUserById(metalakeName, user.id())); + assertFalse(service.deleteUserById(metalakeName, user.id())); + } + private UserMetaService userMetaService() throws IOException { createAndInsertMakeLake(metalakeName); return UserMetaService.getInstance(); } + private MetalakePO getMetalakePO() { + return SessionUtils.getWithoutCommit( + MetalakeMetaMapper.class, mapper -> mapper.selectMetalakeMetaByName(metalakeName)); + } + + private UserPO getUserPO(String userName) { + MetalakePO metalakePO = getMetalakePO(); + return SessionUtils.getWithoutCommit( + UserMetaMapper.class, + mapper -> mapper.selectUserMetaByMetalakeIdAndName(metalakePO.getMetalakeId(), userName)); + } + private void assertThrowsExt(Class type, Executable executable) { Assertions.assertThrows(type, executable); } @@ -1441,4 +1630,19 @@ private Integer countUserRoleRels() { } return count; } + + private void advanceUserVersion(long userId) { + try (SqlSession sqlSession = + SqlSessionFactoryHelper.getInstance().getSqlSessionFactory().openSession(true); + Connection connection = sqlSession.getConnection(); + Statement statement = connection.createStatement()) { + assertEquals( + 1, + statement.executeUpdate( + "UPDATE user_meta SET current_version = current_version + 1 WHERE user_id = " + + userId)); + } catch (SQLException e) { + throw new RuntimeException("Advance user version failed", e); + } + } } 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 bff96b2bd60..a5435ba157f 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 @@ -51,6 +51,7 @@ import org.apache.gravitino.meta.CatalogEntity; import org.apache.gravitino.meta.ColumnEntity; import org.apache.gravitino.meta.FilesetEntity; +import org.apache.gravitino.meta.GroupEntity; import org.apache.gravitino.meta.ModelEntity; import org.apache.gravitino.meta.ModelVersionEntity; import org.apache.gravitino.meta.PolicyEntity; @@ -61,6 +62,7 @@ import org.apache.gravitino.meta.TableStatisticEntity; import org.apache.gravitino.meta.TagEntity; import org.apache.gravitino.meta.TopicEntity; +import org.apache.gravitino.meta.UserEntity; import org.apache.gravitino.policy.Policy; import org.apache.gravitino.policy.PolicyContent; import org.apache.gravitino.policy.PolicyContents; @@ -79,6 +81,7 @@ import org.apache.gravitino.storage.relational.po.ColumnPO; import org.apache.gravitino.storage.relational.po.FilesetPO; import org.apache.gravitino.storage.relational.po.FilesetVersionPO; +import org.apache.gravitino.storage.relational.po.GroupPO; import org.apache.gravitino.storage.relational.po.MetalakePO; import org.apache.gravitino.storage.relational.po.ModelPO; import org.apache.gravitino.storage.relational.po.ModelVersionAliasRelPO; @@ -93,6 +96,7 @@ import org.apache.gravitino.storage.relational.po.TagMetadataObjectRelPO; import org.apache.gravitino.storage.relational.po.TagPO; import org.apache.gravitino.storage.relational.po.TopicPO; +import org.apache.gravitino.storage.relational.po.UserPO; import org.apache.gravitino.utils.NameIdentifierUtil; import org.apache.gravitino.utils.NamespaceUtil; import org.junit.jupiter.api.Assertions; @@ -665,6 +669,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()); } @@ -679,6 +685,8 @@ public void testUpdateCatalogPOVersion() { 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.getCatalogComment()); } @@ -696,6 +704,8 @@ public void testUpdateSchemaPOVersion() { 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.getSchemaComment()); } @@ -795,6 +805,28 @@ public void testUpdateFilesetPOVersion() throws JsonProcessingException { assertEquals("test1", updatePO2.getFilesetName()); } + @Test + public void testUpdatePrincipalPOVersion() { + AuditInfo auditInfo = + AuditInfo.builder().withCreator("creator").withCreateTime(FIX_INSTANT).build(); + UserEntity user = + UserEntity.builder().withId(1L).withName("user").withAuditInfo(auditInfo).build(); + GroupEntity group = + GroupEntity.builder().withId(2L).withName("group").withAuditInfo(auditInfo).build(); + UserPO userPO = + POConverters.initializeUserPOWithVersion(user, UserPO.builder().withMetalakeId(1L)); + GroupPO groupPO = + POConverters.initializeGroupPOWithVersion(group, GroupPO.builder().withMetalakeId(1L)); + + UserPO updatedUserPO = POConverters.updateUserPOWithVersion(userPO, user); + GroupPO updatedGroupPO = POConverters.updateGroupPOWithVersion(groupPO, group); + + assertEquals(2, updatedUserPO.getCurrentVersion()); + assertEquals(2, updatedUserPO.getLastVersion()); + assertEquals(2, updatedGroupPO.getCurrentVersion()); + assertEquals(2, updatedGroupPO.getLastVersion()); + } + @Test public void testFromPolicyPO() throws JsonProcessingException { ImmutableSet supportedObjectTypes =