yuqi1129 commented on code in PR #12350:
URL: https://github.com/apache/gravitino/pull/12350#discussion_r3765795548
##########
core/src/main/java/org/apache/gravitino/storage/relational/service/MetalakeMetaService.java:
##########
@@ -420,6 +423,75 @@ public boolean deleteMetalake(NameIdentifier ident,
boolean cascade) {
return true;
}
+ void deleteMetalakeWithVersion(NameIdentifier identifier, Long metalakeId,
Long currentVersion) {
+ int deleted =
+ SessionUtils.getWithoutCommit(
+ MetalakeMetaMapper.class,
+ mapper -> mapper.softDeleteMetalakeMetaByMetalakeId(metalakeId,
currentVersion));
+ if (deleted == 0) {
+ throw metalakeWriteFailure(identifier, metalakeId, identifier.name());
+ }
+ }
+
+ private RuntimeException metalakeWriteFailure(
+ NameIdentifier identifier, Long metalakeId, String observedName) {
+ MetalakePO currentMetalakePO =
+ SessionUtils.getWithoutCommit(
+ MetalakeMetaMapper.class, mapper ->
mapper.selectMetalakeMetaByIdForUpdate(metalakeId));
+ if (currentMetalakePO == null
+ || !Objects.equals(currentMetalakePO.getMetalakeName(), observedName))
{
+ return new NoSuchEntityException(
+ NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
+ Entity.EntityType.METALAKE.name().toLowerCase(),
+ identifier.name());
+ }
+ return optimisticLockException(identifier);
+ }
+
+ private void deleteCatalogsWithVersions(NameIdentifier metalakeIdentifier,
Long metalakeId) {
+ List<CatalogPO> catalogPOs =
+ SessionUtils.getWithoutCommit(
+ CatalogMetaMapper.class,
+ mapper -> mapper.listCatalogPOsByMetalakeIdForUpdate(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);
+ }
+ }
+
+ List<SchemaPO> listSchemaPOsForCascade(Long metalakeId) {
+ return SessionUtils.getWithoutCommit(
+ SchemaMetaMapper.class, mapper ->
mapper.listSchemaPOsByMetalakeId(metalakeId));
+ }
+
+ private void deleteSchemasWithVersions(
+ NameIdentifier metalakeIdentifier, List<SchemaPO> schemaPOs) {
+ if (schemaPOs.isEmpty()) {
+ return;
+ }
+ int deleted =
+ SessionUtils.getWithoutCommit(
+ SchemaMetaMapper.class, mapper ->
mapper.softDeleteSchemaMetasWithVersion(schemaPOs));
+ if (deleted != schemaPOs.size()) {
+ throw new OptimisticLockException(
+ "A schema 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);
+ }
Review Comment:
Strictly, it's not duplicated code and is only used for the `metalake`.
Anyway, we need to refine it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
