jerryshao commented on code in PR #12350:
URL: https://github.com/apache/gravitino/pull/12350#discussion_r3765742512
##########
core/src/main/java/org/apache/gravitino/storage/relational/service/CatalogMetaService.java:
##########
@@ -415,6 +420,72 @@ public boolean deleteCatalog(NameIdentifier identifier,
boolean cascade) {
return true;
}
+ private void deleteCatalogWithVersion(NameIdentifier identifier, CatalogPO
observedCatalogPO) {
+ int deleted =
+ SessionUtils.getWithoutCommit(
+ CatalogMetaMapper.class,
+ mapper ->
+ mapper.softDeleteCatalogMetasByCatalogId(
+ observedCatalogPO.getCatalogId(),
observedCatalogPO.getCurrentVersion()));
+ if (deleted == 0) {
+ throw catalogWriteFailure(identifier, observedCatalogPO);
+ }
+ }
+
+ private void lockMetalakeForCatalogCreate(MetalakePO observedMetalakePO) {
+ MetalakePO currentMetalakePO =
+ SessionUtils.getWithoutCommit(
+ MetalakeMetaMapper.class,
+ mapper ->
mapper.selectMetalakeMetaByIdForShare(observedMetalakePO.getMetalakeId()));
+ if (currentMetalakePO == null
+ || !Objects.equals(
+ currentMetalakePO.getMetalakeName(),
observedMetalakePO.getMetalakeName())) {
+ throw new NoSuchEntityException(
+ NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
+ Entity.EntityType.METALAKE.name().toLowerCase(),
+ observedMetalakePO.getMetalakeName());
+ }
+ }
+
+ private RuntimeException catalogWriteFailure(
+ NameIdentifier identifier, CatalogPO observedCatalogPO) {
+ CatalogPO currentCatalogPO =
+ SessionUtils.getWithoutCommit(
+ CatalogMetaMapper.class,
+ mapper ->
mapper.selectCatalogMetaByIdForUpdate(observedCatalogPO.getCatalogId()));
+ if (currentCatalogPO == null
+ || !Objects.equals(currentCatalogPO.getCatalogName(),
observedCatalogPO.getCatalogName())
+ || !Objects.equals(currentCatalogPO.getMetalakeId(),
observedCatalogPO.getMetalakeId())) {
+ return new NoSuchEntityException(
+ NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
+ Entity.EntityType.CATALOG.name().toLowerCase(),
+ identifier.name());
+ }
+ return optimisticLockException(identifier);
+ }
+
+ private void deleteSchemasWithVersions(NameIdentifier catalogIdentifier,
Long catalogId) {
+ List<SchemaPO> 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);
+ }
Review Comment:
We can define this method in the helper method.
--
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]
