yuqi1129 commented on code in PR #12454:
URL: https://github.com/apache/gravitino/pull/12454#discussion_r3793420810
##########
core/src/main/java/org/apache/gravitino/storage/relational/service/MetalakeMetaService.java:
##########
@@ -383,6 +385,74 @@ 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) {
+ // This re-read is deliberately a locking read. Under MySQL REPEATABLE
READ a plain SELECT
+ // returns this transaction's snapshot, which still shows a row that a
concurrent writer has
+ // already deleted or renamed away, so a stale-version conflict and a
missing entity would be
+ // indistinguishable. A locking read observes the latest committed row
instead. It costs no
+ // extra waiting in practice: the compare-and-set above is an UPDATE that
already queued on the
+ // same row lock, so the competing writer has committed by the time
control reaches here.
+ MetalakePO currentMetalakePO =
+ SessionUtils.getWithoutCommit(
+ MetalakeMetaMapper.class, mapper ->
mapper.selectMetalakeMetaByIdForUpdate(metalakeId));
+ if (currentMetalakePO == null
+ || !Objects.equals(currentMetalakePO.getMetalakeName(), observedName))
{
+ return new NoSuchEntityException(
+ NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
+ Entity.EntityType.METALAKE.name().toLowerCase(),
+ identifier.name());
+ }
+ return ExceptionUtils.concurrentModification(Entity.EntityType.METALAKE,
identifier);
+ }
+
+ private void deleteCatalogsWithVersions(NameIdentifier metalakeIdentifier,
Long metalakeId) {
+ 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 ExceptionUtils.concurrentChildModification(
+ Entity.EntityType.CATALOG, Entity.EntityType.METALAKE,
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 ExceptionUtils.concurrentChildModification(
+ Entity.EntityType.SCHEMA, Entity.EntityType.METALAKE,
metalakeIdentifier);
+ }
+ }
+
Review Comment:
Indeed. I did add some operations like DB read/write locks (for share/for
write) and extra checks. For example, we need to keep checking for emptiness
when removing metalake/catalog and so on.
--
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]
