This is an automated email from the ASF dual-hosted git repository.
yuqi1129 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new 54567b407f [#12639] refactor(core): Extract shared OCC helpers into
OccWriteSupport (#12666)
54567b407f is described below
commit 54567b407f1d0fd0c5932f559bc47dad667b0782
Author: Tanay Paul <[email protected]>
AuthorDate: Tue Sep 1 08:31:14 2026 +0530
[#12639] refactor(core): Extract shared OCC helpers into OccWriteSupport
(#12666)
### What changes were proposed in this pull request?
Introduce `OccWriteSupport`, a utility class with four static generic
helpers that cover the OCC patterns duplicated across relational meta
services:
- `writeFailure` - locking reread by id, compare natural-key fields,
return `NoSuchEntityException` or `OptimisticLockException`
- `deleteWithVersion` - single-row CAS soft delete guarded by version
- `deleteChildrenWithVersions` - batch CAS soft delete of children,
fail-fast on count mismatch
- `lockParentForChildWrite` - shared/exclusive parent row lock before a
child write or delete
Migrate MetalakeMetaService, CatalogMetaService, SchemaMetaService, and
TableMetaService to the shared helpers. No behavior change.
### Why are the changes needed?
Each OCC PR copied the previous service's helpers. A future change to
classification rules (extra field, different lock semantics) or a new
entity type would need the same edit in four places.
Fix: #12639
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Added `TestOccWriteSupport` (11 tests covering all helpers and edge
cases). Ran the full relational service test suite: `./gradlew
:core:test --tests "org.apache.gravitino.storage.relational.service.*"
-PskipDockerTests=true` - 266 tests, 0 failures.
---
.../relational/service/CatalogMetaService.java | 88 ++++-----
.../relational/service/MetalakeMetaService.java | 81 ++++-----
.../relational/service/OccWriteSupport.java | 152 ++++++++++++++++
.../relational/service/SchemaMetaService.java | 154 +++++++---------
.../relational/service/TableMetaService.java | 54 +++---
.../relational/service/TestOccWriteSupport.java | 200 +++++++++++++++++++++
6 files changed, 511 insertions(+), 218 deletions(-)
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 8f32ac0563..66970fd4cc 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
@@ -435,15 +435,14 @@ public class CatalogMetaService {
* loses the race to another writer must not delete a catalog it never saw.
*/
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);
- }
+ OccWriteSupport.deleteWithVersion(
+ () ->
+ SessionUtils.getWithoutCommit(
+ CatalogMetaMapper.class,
+ mapper ->
+ mapper.softDeleteCatalogMetasByCatalogId(
+ observedCatalogPO.getCatalogId(),
observedCatalogPO.getCurrentVersion())),
+ () -> catalogWriteFailure(identifier, observedCatalogPO));
}
/**
@@ -460,18 +459,16 @@ public class CatalogMetaService {
* exists.
*/
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());
- }
+ OccWriteSupport.lockParentForChildWrite(
+ observedMetalakePO.getMetalakeName(),
+ Entity.EntityType.METALAKE,
+ () ->
+ SessionUtils.getWithoutCommit(
+ MetalakeMetaMapper.class,
+ mapper ->
+
mapper.selectMetalakeMetaByIdForShare(observedMetalakePO.getMetalakeId())),
+ null,
+ current -> Objects.equals(current.getMetalakeName(),
observedMetalakePO.getMetalakeName()));
}
/**
@@ -481,23 +478,17 @@ public class CatalogMetaService {
*/
private RuntimeException catalogWriteFailure(
NameIdentifier identifier, CatalogPO observedCatalogPO) {
- // Sessions run at READ_COMMITTED, so a plain read would already see the
latest committed row.
- // The locking read additionally waits for a writer that is still in
flight, so a rename or
- // delete that has not committed yet is classified as not-found instead of
as a stale-version
- // conflict. The lock is taken on the error path of a transaction that is
about to roll back.
- 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 ExceptionUtils.concurrentModification(Entity.EntityType.CATALOG,
identifier);
+ return OccWriteSupport.writeFailure(
+ identifier,
+ Entity.EntityType.CATALOG,
+ () ->
+ SessionUtils.getWithoutCommit(
+ CatalogMetaMapper.class,
+ mapper ->
mapper.selectCatalogMetaByIdForUpdate(observedCatalogPO.getCatalogId())),
+ null,
+ current ->
+ Objects.equals(current.getCatalogName(),
observedCatalogPO.getCatalogName())
+ && Objects.equals(current.getMetalakeId(),
observedCatalogPO.getMetalakeId()));
}
/**
@@ -506,18 +497,15 @@ public class CatalogMetaService {
*/
private void deleteSchemasWithVersions(NameIdentifier catalogIdentifier,
Long catalogId) {
List<SchemaPO> schemaPOs = listSchemaPOsForCascade(catalogId);
- if (schemaPOs.isEmpty()) {
- return;
- }
- int deleted =
- SessionUtils.getWithoutCommit(
- SchemaMetaMapper.class, mapper ->
mapper.softDeleteSchemaMetasWithVersion(schemaPOs));
- // A smaller count means one of these schemas was altered by someone who
did not take the
- // catalog row lock. Never commit half a cascade: roll the whole
transaction back instead.
- if (deleted != schemaPOs.size()) {
- throw ExceptionUtils.concurrentChildModification(
- Entity.EntityType.SCHEMA, Entity.EntityType.CATALOG,
catalogIdentifier);
- }
+ OccWriteSupport.deleteChildrenWithVersions(
+ catalogIdentifier,
+ Entity.EntityType.SCHEMA,
+ Entity.EntityType.CATALOG,
+ schemaPOs,
+ children ->
+ SessionUtils.getWithoutCommit(
+ SchemaMetaMapper.class,
+ mapper -> mapper.softDeleteSchemaMetasWithVersion(children)));
}
/**
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 fb63973758..e7cf7a9cf5 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
@@ -403,33 +403,25 @@ public class MetalakeMetaService {
}
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());
- }
+ OccWriteSupport.deleteWithVersion(
+ () ->
+ SessionUtils.getWithoutCommit(
+ MetalakeMetaMapper.class,
+ mapper ->
mapper.softDeleteMetalakeMetaByMetalakeId(metalakeId, currentVersion)),
+ () -> metalakeWriteFailure(identifier, metalakeId, identifier.name()));
}
private RuntimeException metalakeWriteFailure(
NameIdentifier identifier, Long metalakeId, String observedName) {
- // Sessions run at READ_COMMITTED, so a plain read would already see the
latest committed row.
- // The locking read additionally waits for a writer that is still in
flight, so a delete or
- // rename that has not committed yet is reported as a missing metalake
instead of as a stale
- // version conflict. The lock is taken on the error path of a transaction
that is about to roll
- // back.
- 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);
+ return OccWriteSupport.writeFailure(
+ identifier,
+ Entity.EntityType.METALAKE,
+ () ->
+ SessionUtils.getWithoutCommit(
+ MetalakeMetaMapper.class,
+ mapper -> mapper.selectMetalakeMetaByIdForUpdate(metalakeId)),
+ null,
+ current -> Objects.equals(current.getMetalakeName(), observedName));
}
private void deleteCatalogsWithVersions(NameIdentifier metalakeIdentifier,
Long metalakeId) {
@@ -441,19 +433,15 @@ public class MetalakeMetaService {
SessionUtils.getWithoutCommit(
CatalogMetaMapper.class,
mapper -> mapper.listCatalogPOsByMetalakeIdForUpdate(metalakeId));
- if (catalogPOs.isEmpty()) {
- return;
- }
- int deleted =
- SessionUtils.getWithoutCommit(
- CatalogMetaMapper.class,
- mapper -> mapper.softDeleteCatalogMetasWithVersion(catalogPOs));
- // Never commit a partial cascade. A smaller count means that a catalog no
longer matches the
- // ID and version read above, so the outer transaction must roll back all
deletes.
- if (deleted != catalogPOs.size()) {
- throw ExceptionUtils.concurrentChildModification(
- Entity.EntityType.CATALOG, Entity.EntityType.METALAKE,
metalakeIdentifier);
- }
+ OccWriteSupport.deleteChildrenWithVersions(
+ metalakeIdentifier,
+ Entity.EntityType.CATALOG,
+ Entity.EntityType.METALAKE,
+ catalogPOs,
+ children ->
+ SessionUtils.getWithoutCommit(
+ CatalogMetaMapper.class,
+ mapper -> mapper.softDeleteCatalogMetasWithVersion(children)));
}
List<SchemaPO> listSchemaPOsForCascade(Long metalakeId) {
@@ -463,18 +451,15 @@ public class MetalakeMetaService {
private void deleteSchemasWithVersions(
NameIdentifier metalakeIdentifier, List<SchemaPO> schemaPOs) {
- if (schemaPOs.isEmpty()) {
- return;
- }
- int deleted =
- SessionUtils.getWithoutCommit(
- SchemaMetaMapper.class, mapper ->
mapper.softDeleteSchemaMetasWithVersion(schemaPOs));
- // The version check protects this snapshot from a schema alter that does
not use the parent
- // catalog lock. Roll back the whole cascade instead of silently losing
that schema change.
- if (deleted != schemaPOs.size()) {
- throw ExceptionUtils.concurrentChildModification(
- Entity.EntityType.SCHEMA, Entity.EntityType.METALAKE,
metalakeIdentifier);
- }
+ OccWriteSupport.deleteChildrenWithVersions(
+ metalakeIdentifier,
+ Entity.EntityType.SCHEMA,
+ Entity.EntityType.METALAKE,
+ schemaPOs,
+ children ->
+ SessionUtils.getWithoutCommit(
+ SchemaMetaMapper.class,
+ mapper -> mapper.softDeleteSchemaMetasWithVersion(children)));
}
@Monitored(
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/OccWriteSupport.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/OccWriteSupport.java
new file mode 100644
index 0000000000..3e5004a896
--- /dev/null
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/OccWriteSupport.java
@@ -0,0 +1,152 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.storage.relational.service;
+
+import java.util.List;
+import java.util.Locale;
+import java.util.function.Function;
+import java.util.function.IntSupplier;
+import java.util.function.Predicate;
+import java.util.function.Supplier;
+import java.util.function.ToIntFunction;
+import javax.annotation.Nullable;
+import org.apache.gravitino.Entity;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.exceptions.NoSuchEntityException;
+import org.apache.gravitino.storage.relational.utils.ExceptionUtils;
+
+/**
+ * Utility class providing shared helpers for optimistic concurrency control
(OCC) operations across
+ * relational metadata services.
+ */
+public class OccWriteSupport {
+
+ private OccWriteSupport() {}
+
+ /**
+ * Classifies a write-failure for an entity during an optimistic concurrency
control operation.
+ *
+ * <p>Executes a locking lookup to re-read the target entity. If the entity
no longer exists or
+ * its natural key fields do not match the expected identity, returns a
{@link
+ * NoSuchEntityException}. Otherwise, returns an {@link
+ * org.apache.gravitino.exceptions.OptimisticLockException} via {@link
+ * ExceptionUtils#concurrentModification(Entity.EntityType, NameIdentifier)}.
+ *
+ * @param <T> the persistent object (PO) type of the entity
+ * @param identifier the name identifier of the entity
+ * @param type the entity type
+ * @param lockingLookup a supplier that retrieves the current entity while
locking its row
+ * @param poMapper an optional function to transform the retrieved PO (e.g.
physical to logical)
+ * @param sameIdentity a predicate comparing the retrieved PO against
expected natural key values
+ * @return the classified RuntimeException to be thrown
+ */
+ public static <T> RuntimeException writeFailure(
+ NameIdentifier identifier,
+ Entity.EntityType type,
+ Supplier<T> lockingLookup,
+ @Nullable Function<T, T> poMapper,
+ @Nullable Predicate<T> sameIdentity) {
+ T currentPO = lockingLookup.get();
+ if (currentPO != null && poMapper != null) {
+ currentPO = poMapper.apply(currentPO);
+ }
+ if (currentPO == null || (sameIdentity != null &&
!sameIdentity.test(currentPO))) {
+ return new NoSuchEntityException(
+ NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
+ type.name().toLowerCase(Locale.ROOT),
+ identifier.name());
+ }
+ return ExceptionUtils.concurrentModification(type, identifier);
+ }
+
+ /**
+ * Executes a single-row compare-and-set soft delete for an entity guarded
by version.
+ *
+ * @param softDeleteOps an operation supplying the number of rows affected
by soft delete
+ * @param onMissSupplier a supplier providing the RuntimeException when zero
rows are deleted
+ */
+ public static void deleteWithVersion(
+ IntSupplier softDeleteOps, Supplier<RuntimeException> onMissSupplier) {
+ int deleted = softDeleteOps.getAsInt();
+ if (deleted == 0) {
+ throw onMissSupplier.get();
+ }
+ }
+
+ /**
+ * Executes a batch soft delete of child entities guarded by their
individual versions.
+ *
+ * <p>If the affected row count does not match the size of the child list,
throws a concurrent
+ * child modification exception via {@link
ExceptionUtils#concurrentChildModification}.
+ *
+ * @param <T> the persistent object (PO) type of the child entities
+ * @param parentIdentifier the name identifier of the parent entity
+ * @param childType the entity type of the children
+ * @param parentType the entity type of the parent
+ * @param children the list of child persistent objects to delete
+ * @param softDeleteOps a function performing batch soft delete on the child
list
+ */
+ public static <T> void deleteChildrenWithVersions(
+ NameIdentifier parentIdentifier,
+ Entity.EntityType childType,
+ Entity.EntityType parentType,
+ @Nullable List<T> children,
+ ToIntFunction<List<T>> softDeleteOps) {
+ if (children == null || children.isEmpty()) {
+ return;
+ }
+ int deleted = softDeleteOps.applyAsInt(children);
+ if (deleted != children.size()) {
+ throw ExceptionUtils.concurrentChildModification(childType, parentType,
parentIdentifier);
+ }
+ }
+
+ /**
+ * Locks the parent row before executing a child write or delete operation
to guarantee existence
+ * and identity consistency.
+ *
+ * @param <P> the persistent object (PO) type of the parent entity
+ * @param parentEntityName the name of the parent entity expected in
exceptions
+ * @param parentType the entity type of the parent
+ * @param lockingLookup a supplier that retrieves the parent entity while
locking its row
+ * @param poMapper an optional function to transform the retrieved parent PO
+ * @param sameParentIdentity a predicate verifying parent identity (e.g.
name and ancestor ids)
+ * @return the locked parent persistent object
+ * @throws NoSuchEntityException if the parent entity is missing or fails
identity validation
+ */
+ public static <P> P lockParentForChildWrite(
+ String parentEntityName,
+ Entity.EntityType parentType,
+ Supplier<P> lockingLookup,
+ @Nullable Function<P, P> poMapper,
+ @Nullable Predicate<P> sameParentIdentity) {
+ P currentParent = lockingLookup.get();
+ if (currentParent != null && poMapper != null) {
+ currentParent = poMapper.apply(currentParent);
+ }
+ if (currentParent == null
+ || (sameParentIdentity != null &&
!sameParentIdentity.test(currentParent))) {
+ throw new NoSuchEntityException(
+ NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
+ parentType.name().toLowerCase(Locale.ROOT),
+ parentEntityName);
+ }
+ return currentParent;
+ }
+}
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 40eacb0b71..037befda7e 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
@@ -404,15 +404,14 @@ public class SchemaMetaService {
* lost the race must not delete a schema it never looked at.
*/
private void deleteSchemaWithVersion(NameIdentifier identifier, SchemaPO
observedSchemaPO) {
- int deleted =
- SessionUtils.getWithoutCommit(
- SchemaMetaMapper.class,
- mapper ->
- mapper.softDeleteSchemaMetaBySchemaIdAndVersion(
- observedSchemaPO.getSchemaId(),
observedSchemaPO.getCurrentVersion()));
- if (deleted == 0) {
- throw schemaWriteFailure(identifier, observedSchemaPO);
- }
+ OccWriteSupport.deleteWithVersion(
+ () ->
+ SessionUtils.getWithoutCommit(
+ SchemaMetaMapper.class,
+ mapper ->
+ mapper.softDeleteSchemaMetaBySchemaIdAndVersion(
+ observedSchemaPO.getSchemaId(),
observedSchemaPO.getCurrentVersion())),
+ () -> schemaWriteFailure(identifier, observedSchemaPO));
}
@Monitored(
@@ -465,21 +464,20 @@ public class SchemaMetaService {
*/
private void lockCatalogForSchemaCreate(
CatalogPO observedCatalogPO, boolean createsImplicitAncestors) {
- CatalogPO currentCatalogPO =
- SessionUtils.getWithoutCommit(
- CatalogMetaMapper.class,
- mapper ->
- createsImplicitAncestors
- ?
mapper.selectCatalogMetaByIdForUpdate(observedCatalogPO.getCatalogId())
- :
mapper.selectCatalogMetaByIdForShare(observedCatalogPO.getCatalogId()));
- if (currentCatalogPO == null
- || !Objects.equals(currentCatalogPO.getCatalogName(),
observedCatalogPO.getCatalogName())
- || !Objects.equals(currentCatalogPO.getMetalakeId(),
observedCatalogPO.getMetalakeId())) {
- throw new NoSuchEntityException(
- NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
- Entity.EntityType.CATALOG.name().toLowerCase(),
- observedCatalogPO.getCatalogName());
- }
+ OccWriteSupport.lockParentForChildWrite(
+ observedCatalogPO.getCatalogName(),
+ Entity.EntityType.CATALOG,
+ () ->
+ SessionUtils.getWithoutCommit(
+ CatalogMetaMapper.class,
+ mapper ->
+ createsImplicitAncestors
+ ?
mapper.selectCatalogMetaByIdForUpdate(observedCatalogPO.getCatalogId())
+ :
mapper.selectCatalogMetaByIdForShare(observedCatalogPO.getCatalogId())),
+ null,
+ current ->
+ Objects.equals(current.getCatalogName(),
observedCatalogPO.getCatalogName())
+ && Objects.equals(current.getMetalakeId(),
observedCatalogPO.getMetalakeId()));
}
/**
@@ -489,18 +487,18 @@ public class SchemaMetaService {
* two overlapping cascades cannot deadlock.
*/
private void lockCatalogForSchemaDelete(NameIdentifier identifier, SchemaPO
observedSchemaPO) {
- CatalogPO currentCatalogPO =
- SessionUtils.getWithoutCommit(
- CatalogMetaMapper.class,
- mapper ->
mapper.selectCatalogMetaByIdForUpdate(observedSchemaPO.getCatalogId()));
- if (currentCatalogPO == null
- || !Objects.equals(currentCatalogPO.getCatalogName(),
identifier.namespace().level(1))
- || !Objects.equals(currentCatalogPO.getMetalakeId(),
observedSchemaPO.getMetalakeId())) {
- throw new NoSuchEntityException(
- NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
- Entity.EntityType.CATALOG.name().toLowerCase(),
- identifier.namespace().level(1));
- }
+ String catalogName = identifier.namespace().level(1);
+ OccWriteSupport.lockParentForChildWrite(
+ catalogName,
+ Entity.EntityType.CATALOG,
+ () ->
+ SessionUtils.getWithoutCommit(
+ CatalogMetaMapper.class,
+ mapper ->
mapper.selectCatalogMetaByIdForUpdate(observedSchemaPO.getCatalogId())),
+ null,
+ current ->
+ Objects.equals(current.getCatalogName(), catalogName)
+ && Objects.equals(current.getMetalakeId(),
observedSchemaPO.getMetalakeId()));
}
/**
@@ -515,19 +513,18 @@ public class SchemaMetaService {
Long observedCatalogId,
Long observedMetalakeId) {
NameIdentifier schemaIdentifier =
NameIdentifierUtil.getSchemaIdentifier(entityIdentifier);
- SchemaPO currentSchemaPO =
- SessionUtils.getWithoutCommit(
- SchemaMetaMapper.class,
- mapper -> mapper.selectSchemaMetaByIdForShare(observedSchemaId));
- if (currentSchemaPO != null) {
- currentSchemaPO = physicalToLogicalSchemaPO(currentSchemaPO);
- }
- if (currentSchemaPO == null
- || !Objects.equals(currentSchemaPO.getSchemaName(),
schemaIdentifier.name())
- || !Objects.equals(currentSchemaPO.getCatalogId(), observedCatalogId)
- || !Objects.equals(currentSchemaPO.getMetalakeId(),
observedMetalakeId)) {
- throw noSuchSchemaException(schemaIdentifier);
- }
+ OccWriteSupport.lockParentForChildWrite(
+ schemaIdentifier.name(),
+ Entity.EntityType.SCHEMA,
+ () ->
+ SessionUtils.getWithoutCommit(
+ SchemaMetaMapper.class,
+ mapper ->
mapper.selectSchemaMetaByIdForShare(observedSchemaId)),
+ SchemaMetaService::physicalToLogicalSchemaPO,
+ current ->
+ Objects.equals(current.getSchemaName(), schemaIdentifier.name())
+ && Objects.equals(current.getCatalogId(), observedCatalogId)
+ && Objects.equals(current.getMetalakeId(),
observedMetalakeId));
}
/**
@@ -537,32 +534,18 @@ public class SchemaMetaService {
*/
private RuntimeException schemaWriteFailure(
NameIdentifier identifier, SchemaPO observedSchemaPO) {
- // Sessions run at READ_COMMITTED, so a plain read would already see the
latest committed row.
- // The locking read additionally waits for a writer that is still in
flight, so a delete or
- // rename that has not committed yet is reported as a missing schema
instead of as a stale
- // version conflict. The lock is taken on the error path of a transaction
that is about to roll
- // back.
- SchemaPO currentSchemaPO =
- SessionUtils.getWithoutCommit(
- SchemaMetaMapper.class,
- mapper ->
mapper.selectSchemaMetaByIdForUpdate(observedSchemaPO.getSchemaId()));
- if (currentSchemaPO == null) {
- return noSuchSchemaException(identifier);
- }
- currentSchemaPO = physicalToLogicalSchemaPO(currentSchemaPO);
- if (!Objects.equals(currentSchemaPO.getSchemaName(),
observedSchemaPO.getSchemaName())
- || !Objects.equals(currentSchemaPO.getCatalogId(),
observedSchemaPO.getCatalogId())
- || !Objects.equals(currentSchemaPO.getMetalakeId(),
observedSchemaPO.getMetalakeId())) {
- return noSuchSchemaException(identifier);
- }
- return ExceptionUtils.concurrentModification(Entity.EntityType.SCHEMA,
identifier);
- }
-
- private NoSuchEntityException noSuchSchemaException(NameIdentifier
identifier) {
- return new NoSuchEntityException(
- NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
- Entity.EntityType.SCHEMA.name().toLowerCase(),
- identifier.name());
+ return OccWriteSupport.writeFailure(
+ identifier,
+ Entity.EntityType.SCHEMA,
+ () ->
+ SessionUtils.getWithoutCommit(
+ SchemaMetaMapper.class,
+ mapper ->
mapper.selectSchemaMetaByIdForUpdate(observedSchemaPO.getSchemaId())),
+ SchemaMetaService::physicalToLogicalSchemaPO,
+ current ->
+ Objects.equals(current.getSchemaName(),
observedSchemaPO.getSchemaName())
+ && Objects.equals(current.getCatalogId(),
observedSchemaPO.getCatalogId())
+ && Objects.equals(current.getMetalakeId(),
observedSchemaPO.getMetalakeId()));
}
/**
@@ -571,18 +554,15 @@ public class SchemaMetaService {
*/
private void deleteDescendantSchemasWithVersions(
NameIdentifier schemaIdentifier, List<SchemaPO> descendants) {
- if (descendants.isEmpty()) {
- return;
- }
- int deleted =
- SessionUtils.getWithoutCommit(
- SchemaMetaMapper.class, mapper ->
mapper.softDeleteSchemaMetasWithVersion(descendants));
- // A smaller count means one of these schemas was altered by a request
that did not take the
- // catalog lock. Never commit half a cascade: roll the whole transaction
back instead.
- if (deleted != descendants.size()) {
- throw ExceptionUtils.concurrentChildModification(
- Entity.EntityType.SCHEMA, Entity.EntityType.SCHEMA,
schemaIdentifier);
- }
+ OccWriteSupport.deleteChildrenWithVersions(
+ schemaIdentifier,
+ Entity.EntityType.SCHEMA,
+ Entity.EntityType.SCHEMA,
+ descendants,
+ children ->
+ SessionUtils.getWithoutCommit(
+ SchemaMetaMapper.class,
+ mapper -> mapper.softDeleteSchemaMetasWithVersion(children)));
}
/**
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/TableMetaService.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/TableMetaService.java
index a387bc370f..5ad3569fd0 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/service/TableMetaService.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/TableMetaService.java
@@ -298,15 +298,14 @@ public class TableMetaService {
* snapshot without copying the production CAS logic.
*/
void deleteTableWithVersion(NameIdentifier identifier, TablePO
observedTablePO) {
- int deleted =
- SessionUtils.getWithoutCommit(
- TableMetaMapper.class,
- mapper ->
- mapper.softDeleteTableMetasByTableId(
- observedTablePO.getTableId(),
observedTablePO.getCurrentVersion()));
- if (deleted == 0) {
- throw tableWriteFailure(identifier, observedTablePO);
- }
+ OccWriteSupport.deleteWithVersion(
+ () ->
+ SessionUtils.getWithoutCommit(
+ TableMetaMapper.class,
+ mapper ->
+ mapper.softDeleteTableMetasByTableId(
+ observedTablePO.getTableId(),
observedTablePO.getCurrentVersion())),
+ () -> tableWriteFailure(identifier, observedTablePO));
}
@Monitored(
@@ -432,29 +431,18 @@ public class TableMetaService {
}
private RuntimeException tableWriteFailure(NameIdentifier identifier,
TablePO observedTablePO) {
- // A zero-row CAS has two different meanings:
- // 1. The same table is still here, but another writer changed its
version. The caller should
- // retry, so return OptimisticLockException.
- // 2. The table ID was deleted, renamed, or moved away from the requested
name. From the
- // caller's point of view the requested table no longer exists, so
return NoSuchEntity.
- //
- // Read by the stable table ID and lock the row. The lock waits for an
in-flight writer to
- // finish, which lets us classify the failure using committed data instead
of guessing while
- // the other transaction is still running.
- TablePO currentTablePO =
- SessionUtils.getWithoutCommit(
- TableMetaMapper.class,
- mapper ->
mapper.selectTableMetaByIdForUpdate(observedTablePO.getTableId()));
- if (currentTablePO == null
- || !Objects.equals(currentTablePO.getTableName(),
observedTablePO.getTableName())
- || !Objects.equals(currentTablePO.getSchemaId(),
observedTablePO.getSchemaId())
- || !Objects.equals(currentTablePO.getCatalogId(),
observedTablePO.getCatalogId())
- || !Objects.equals(currentTablePO.getMetalakeId(),
observedTablePO.getMetalakeId())) {
- return new NoSuchEntityException(
- NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
- Entity.EntityType.TABLE.name().toLowerCase(),
- identifier.name());
- }
- return ExceptionUtils.concurrentModification(Entity.EntityType.TABLE,
identifier);
+ return OccWriteSupport.writeFailure(
+ identifier,
+ Entity.EntityType.TABLE,
+ () ->
+ SessionUtils.getWithoutCommit(
+ TableMetaMapper.class,
+ mapper ->
mapper.selectTableMetaByIdForUpdate(observedTablePO.getTableId())),
+ null,
+ current ->
+ Objects.equals(current.getTableName(),
observedTablePO.getTableName())
+ && Objects.equals(current.getSchemaId(),
observedTablePO.getSchemaId())
+ && Objects.equals(current.getCatalogId(),
observedTablePO.getCatalogId())
+ && Objects.equals(current.getMetalakeId(),
observedTablePO.getMetalakeId()));
}
}
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestOccWriteSupport.java
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestOccWriteSupport.java
new file mode 100644
index 0000000000..fb4a533b95
--- /dev/null
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestOccWriteSupport.java
@@ -0,0 +1,200 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.storage.relational.service;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import org.apache.gravitino.Entity;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.exceptions.NoSuchEntityException;
+import org.apache.gravitino.exceptions.OptimisticLockException;
+import org.junit.jupiter.api.Test;
+
+public class TestOccWriteSupport {
+
+ private static class DummyPO {
+ private final String name;
+ private final Long parentId;
+
+ DummyPO(String name, Long parentId) {
+ this.name = name;
+ this.parentId = parentId;
+ }
+
+ String name() {
+ return name;
+ }
+
+ Long parentId() {
+ return parentId;
+ }
+ }
+
+ @Test
+ void testWriteFailureReturnsNoSuchEntityWhenNotFound() {
+ NameIdentifier ident = NameIdentifier.of("metalake_test");
+ RuntimeException ex =
+ OccWriteSupport.writeFailure(
+ ident, Entity.EntityType.METALAKE, () -> null, null, po -> true);
+ assertInstanceOf(NoSuchEntityException.class, ex);
+ }
+
+ @Test
+ void testWriteFailureReturnsNoSuchEntityWhenIdentityMismatch() {
+ NameIdentifier ident = NameIdentifier.of("metalake_test");
+ DummyPO current = new DummyPO("different_name", 1L);
+ RuntimeException ex =
+ OccWriteSupport.writeFailure(
+ ident,
+ Entity.EntityType.METALAKE,
+ () -> current,
+ null,
+ po -> Objects.equals(po.name(), "metalake_test"));
+ assertInstanceOf(NoSuchEntityException.class, ex);
+ }
+
+ @Test
+ void testWriteFailureReturnsOptimisticLockExceptionWhenMatch() {
+ NameIdentifier ident = NameIdentifier.of("metalake_test");
+ DummyPO current = new DummyPO("metalake_test", 1L);
+ RuntimeException ex =
+ OccWriteSupport.writeFailure(
+ ident,
+ Entity.EntityType.METALAKE,
+ () -> current,
+ null,
+ po -> Objects.equals(po.name(), "metalake_test"));
+ assertInstanceOf(OptimisticLockException.class, ex);
+ }
+
+ @Test
+ void testDeleteWithVersionSuccess() {
+ assertDoesNotThrow(
+ () ->
+ OccWriteSupport.deleteWithVersion(
+ () -> 1, () -> new RuntimeException("Should not be thrown")));
+ }
+
+ @Test
+ void testDeleteWithVersionThrowsOnMiss() {
+ assertThrows(
+ NoSuchEntityException.class,
+ () ->
+ OccWriteSupport.deleteWithVersion(
+ () -> 0,
+ () ->
+ new NoSuchEntityException(
+ NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
"metalake", "test")));
+ }
+
+ @Test
+ void testDeleteChildrenWithVersionsEmptyOrNull() {
+ NameIdentifier parentIdent = NameIdentifier.of("parent");
+ assertDoesNotThrow(
+ () ->
+ OccWriteSupport.deleteChildrenWithVersions(
+ parentIdent,
+ Entity.EntityType.CATALOG,
+ Entity.EntityType.METALAKE,
+ null,
+ list -> 0));
+
+ assertDoesNotThrow(
+ () ->
+ OccWriteSupport.deleteChildrenWithVersions(
+ parentIdent,
+ Entity.EntityType.CATALOG,
+ Entity.EntityType.METALAKE,
+ Collections.emptyList(),
+ list -> 0));
+ }
+
+ @Test
+ void testDeleteChildrenWithVersionsSuccess() {
+ NameIdentifier parentIdent = NameIdentifier.of("parent");
+ List<DummyPO> children = List.of(new DummyPO("c1", 1L), new DummyPO("c2",
1L));
+
+ assertDoesNotThrow(
+ () ->
+ OccWriteSupport.deleteChildrenWithVersions(
+ parentIdent,
+ Entity.EntityType.CATALOG,
+ Entity.EntityType.METALAKE,
+ children,
+ list -> 2));
+ }
+
+ @Test
+ void testDeleteChildrenWithVersionsMismatchThrows() {
+ NameIdentifier parentIdent = NameIdentifier.of("parent");
+ List<DummyPO> children = List.of(new DummyPO("c1", 1L), new DummyPO("c2",
1L));
+
+ assertThrows(
+ OptimisticLockException.class,
+ () ->
+ OccWriteSupport.deleteChildrenWithVersions(
+ parentIdent,
+ Entity.EntityType.CATALOG,
+ Entity.EntityType.METALAKE,
+ children,
+ list -> 1));
+ }
+
+ @Test
+ void testLockParentForChildWriteSuccess() {
+ DummyPO parent = new DummyPO("parent_name", 10L);
+ DummyPO locked =
+ OccWriteSupport.lockParentForChildWrite(
+ "parent_name",
+ Entity.EntityType.METALAKE,
+ () -> parent,
+ null,
+ p -> Objects.equals(p.name(), "parent_name"));
+ assertEquals(parent, locked);
+ }
+
+ @Test
+ void testLockParentForChildWriteNotFoundThrows() {
+ assertThrows(
+ NoSuchEntityException.class,
+ () ->
+ OccWriteSupport.lockParentForChildWrite(
+ "parent_name", Entity.EntityType.METALAKE, () -> null, null, p
-> true));
+ }
+
+ @Test
+ void testLockParentForChildWriteIdentityMismatchThrows() {
+ DummyPO parent = new DummyPO("wrong_parent_name", 10L);
+ assertThrows(
+ NoSuchEntityException.class,
+ () ->
+ OccWriteSupport.lockParentForChildWrite(
+ "parent_name",
+ Entity.EntityType.METALAKE,
+ () -> parent,
+ null,
+ p -> Objects.equals(p.name(), "parent_name")));
+ }
+}
