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 c7449e812b [#12344] improvement(core): add OCC for role writes (#12746)
c7449e812b is described below
commit c7449e812bfd9919309dcc56167e89aaf89c4ad4
Author: Qi Yu <[email protected]>
AuthorDate: Tue Sep 1 16:24:16 2026 +0800
[#12344] improvement(core): add OCC for role writes (#12746)
### What changes were proposed in this pull request?
Add version-CAS optimistic concurrency control and atomic service
operations for managed roles.
- Increment the role version on every update and guard updates and soft
deletes with the observed version.
- Execute the role CAS before privilege, relationship, ownership, and
cleanup mutations in one transaction.
- Fence the parent metalake during role creation and preserve monotonic
versions on overwrite.
- Classify failed writes as either missing entities or optimistic-lock
conflicts.
- Reuse `OccWriteSupport` for role OCC operations.
- Migrate the equivalent duplicated OCC helpers in topic, model,
fileset, user, and group services.
- Preserve model fully qualified names and the existing
fileset/user/group conflict-classification behavior.
### Why are the changes needed?
Concurrent role mutations could otherwise overwrite each other or leave
partially updated privilege and relationship state.
The topic, model, fileset, user, group, and role OCC implementations
also landed around the shared-helper work in #12639, leaving duplicated
write-failure, version-delete, and parent-lock logic that can now use
`OccWriteSupport`.
Fix: #12344
### Does this PR introduce _any_ user-facing change?
Concurrent managed-role writes now report the existing optimistic-lock
conflict response (HTTP 409). No API or configuration keys are changed.
### How was this patch tested?
- `./gradlew :core:spotlessApply`
- `env dockerTest=false ./gradlew :core:test --tests TestRoleMetaService
--tests TestTopicMetaService --tests TestModelMetaService --tests
TestFilesetMetaService --tests TestUserMetaService --tests
TestGroupMetaService --tests TestOccWriteSupport --tests TestAuthMappers
--tests TestPOConverters -PskipITs -PskipDockerTests=true`
- `git diff --check`
- GitHub Backend Integration Test matrix for H2, MySQL, and PostgreSQL.
---
.../storage/relational/mapper/RoleMetaMapper.java | 12 +-
.../mapper/RoleMetaSQLProviderFactory.java | 10 +-
.../provider/base/RoleMetaBaseSQLProvider.java | 25 ++-
.../postgresql/RoleMetaPostgreSQLProvider.java | 15 +-
.../relational/service/FilesetMetaService.java | 41 ++--
.../relational/service/GroupMetaService.java | 65 +++---
.../relational/service/ModelMetaService.java | 46 ++--
.../relational/service/OccWriteSupport.java | 38 +++-
.../relational/service/RoleMetaService.java | 138 +++++++++---
.../relational/service/TopicMetaService.java | 52 +++--
.../relational/service/UserMetaService.java | 69 +++---
.../storage/relational/utils/POConverters.java | 12 +-
.../mapper/provider/base/TestAuthMappers.java | 12 +-
.../relational/service/TestOccWriteSupport.java | 10 +
.../relational/service/TestRoleMetaService.java | 233 +++++++++++++++++++++
.../storage/relational/utils/TestPOConverters.java | 28 +++
16 files changed, 606 insertions(+), 200 deletions(-)
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/RoleMetaMapper.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/RoleMetaMapper.java
index 78fcc921d2..1f03e5df6d 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/RoleMetaMapper.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/RoleMetaMapper.java
@@ -47,6 +47,10 @@ public interface RoleMetaMapper {
RolePO selectRoleMetaByMetalakeIdAndName(
@Param("metalakeId") Long metalakeId, @Param("roleName") String
roleName);
+ /** Returns and locks an active role by ID for the current transaction. */
+ @SelectProvider(type = RoleMetaSQLProviderFactory.class, method =
"selectRoleMetaByIdForUpdate")
+ RolePO selectRoleMetaByIdForUpdate(@Param("roleId") Long roleId);
+
@SelectProvider(
type = RoleMetaSQLProviderFactory.class,
method = "selectRoleIdByMetalakeIdAndName")
@@ -81,8 +85,14 @@ public interface RoleMetaMapper {
Integer updateRoleMeta(
@Param("newRoleMeta") RolePO newRolePO, @Param("oldRoleMeta") RolePO
oldRolePO);
+ /**
+ * Soft-deletes an active role only when its OCC version still matches.
+ *
+ * @return the number of deleted rows
+ */
@UpdateProvider(type = RoleMetaSQLProviderFactory.class, method =
"softDeleteRoleMetaByRoleId")
- void softDeleteRoleMetaByRoleId(@Param("roleId") Long roleId);
+ Integer softDeleteRoleMetaByRoleId(
+ @Param("roleId") Long roleId, @Param("currentVersion") Long
currentVersion);
@UpdateProvider(
type = RoleMetaSQLProviderFactory.class,
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/RoleMetaSQLProviderFactory.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/RoleMetaSQLProviderFactory.java
index 649c897f68..3cc763e1ba 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/RoleMetaSQLProviderFactory.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/RoleMetaSQLProviderFactory.java
@@ -55,6 +55,11 @@ public class RoleMetaSQLProviderFactory {
return getProvider().selectRoleMetaByMetalakeIdAndName(metalakeId,
roleName);
}
+ /** Returns SQL that selects and locks an active role by ID. */
+ public static String selectRoleMetaByIdForUpdate(@Param("roleId") Long
roleId) {
+ return getProvider().selectRoleMetaByIdForUpdate(roleId);
+ }
+
public static String selectRoleIdByMetalakeIdAndName(
@Param("metalakeId") Long metalakeId, @Param("roleName") String name) {
return getProvider().selectRoleIdByMetalakeIdAndName(metalakeId, name);
@@ -90,8 +95,9 @@ public class RoleMetaSQLProviderFactory {
return getProvider().updateRoleMeta(newRolePO, oldRolePO);
}
- public static String softDeleteRoleMetaByRoleId(@Param("roleId") Long
roleId) {
- return getProvider().softDeleteRoleMetaByRoleId(roleId);
+ public static String softDeleteRoleMetaByRoleId(
+ @Param("roleId") Long roleId, @Param("currentVersion") Long
currentVersion) {
+ return getProvider().softDeleteRoleMetaByRoleId(roleId, currentVersion);
}
public static String softDeleteRoleMetasByMetalakeId(@Param("metalakeId")
Long metalakeId) {
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/RoleMetaBaseSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/RoleMetaBaseSQLProvider.java
index d3a7127a32..a63a0682dd 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/RoleMetaBaseSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/RoleMetaBaseSQLProvider.java
@@ -42,6 +42,16 @@ public class RoleMetaBaseSQLProvider {
+ " AND deleted_at = 0";
}
+ /** Returns SQL that selects and locks an active role by ID. */
+ public String selectRoleMetaByIdForUpdate(@Param("roleId") Long roleId) {
+ return "SELECT role_id as roleId, role_name as roleName, metalake_id as
metalakeId,"
+ + " properties, audit_info as auditInfo, current_version as
currentVersion,"
+ + " last_version as lastVersion, deleted_at as deletedAt"
+ + " FROM "
+ + ROLE_TABLE_NAME
+ + " WHERE role_id = #{roleId} AND deleted_at = 0 FOR UPDATE";
+ }
+
public String selectRoleIdByMetalakeIdAndName(
@Param("metalakeId") Long metalakeId, @Param("roleName") String name) {
return "SELECT role_id as roleId FROM "
@@ -147,8 +157,10 @@ public class RoleMetaBaseSQLProvider {
+ " metalake_id = #{roleMeta.metalakeId},"
+ " properties = #{roleMeta.properties},"
+ " audit_info = #{roleMeta.auditInfo},"
- + " current_version = #{roleMeta.currentVersion},"
- + " last_version = #{roleMeta.lastVersion},"
+ // Advance rather than reset the OCC token so a writer holding a
pre-overwrite snapshot
+ // cannot pass a later compare-and-set (an ABA conflict).
+ + " last_version = current_version + 1,"
+ + " current_version = current_version + 1,"
+ " deleted_at = #{roleMeta.deletedAt}";
}
@@ -164,19 +176,18 @@ public class RoleMetaBaseSQLProvider {
+ " last_version = #{newRoleMeta.lastVersion},"
+ " deleted_at = #{newRoleMeta.deletedAt}"
+ " WHERE role_id = #{oldRoleMeta.roleId}"
- + " AND role_name = #{oldRoleMeta.roleName}"
- + " AND metalake_id = #{oldRoleMeta.metalakeId}"
+ " AND current_version = #{oldRoleMeta.currentVersion}"
- + " AND last_version = #{oldRoleMeta.lastVersion}"
+ " AND deleted_at = 0";
}
- public String softDeleteRoleMetaByRoleId(@Param("roleId") Long roleId) {
+ public String softDeleteRoleMetaByRoleId(
+ @Param("roleId") Long roleId, @Param("currentVersion") Long
currentVersion) {
return "UPDATE "
+ ROLE_TABLE_NAME
+ " SET deleted_at = (UNIX_TIMESTAMP() * 1000.0)"
+ " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000"
- + " WHERE role_id = #{roleId} AND deleted_at = 0";
+ + " WHERE role_id = #{roleId}"
+ + " AND current_version = #{currentVersion} AND deleted_at = 0";
}
public String softDeleteRoleMetasByMetalakeId(@Param("metalakeId") Long
metalakeId) {
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/RoleMetaPostgreSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/RoleMetaPostgreSQLProvider.java
index 44de0a3bab..92b8fd8c92 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/RoleMetaPostgreSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/RoleMetaPostgreSQLProvider.java
@@ -26,11 +26,13 @@ import org.apache.ibatis.annotations.Param;
public class RoleMetaPostgreSQLProvider extends RoleMetaBaseSQLProvider {
@Override
- public String softDeleteRoleMetaByRoleId(@Param("roleId") Long roleId) {
+ public String softDeleteRoleMetaByRoleId(
+ @Param("roleId") Long roleId, @Param("currentVersion") Long
currentVersion) {
return "UPDATE "
+ ROLE_TABLE_NAME
+ " SET deleted_at = CAST(EXTRACT(EPOCH FROM CURRENT_TIMESTAMP) * 1000
AS BIGINT)"
- + " WHERE role_id = #{roleId} AND deleted_at = 0";
+ + " WHERE role_id = #{roleId}"
+ + " AND current_version = #{currentVersion} AND deleted_at = 0";
}
@Override
@@ -62,8 +64,13 @@ public class RoleMetaPostgreSQLProvider extends
RoleMetaBaseSQLProvider {
+ " metalake_id = #{roleMeta.metalakeId},"
+ " properties = #{roleMeta.properties},"
+ " audit_info = #{roleMeta.auditInfo},"
- + " current_version = #{roleMeta.currentVersion},"
- + " last_version = #{roleMeta.lastVersion},"
+ // PostgreSQL requires the stored-row column to be qualified in ON
CONFLICT assignments.
+ + " current_version = "
+ + ROLE_TABLE_NAME
+ + ".current_version + 1,"
+ + " last_version = "
+ + ROLE_TABLE_NAME
+ + ".current_version + 1,"
+ " deleted_at = #{roleMeta.deletedAt}";
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/FilesetMetaService.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/FilesetMetaService.java
index 70ce93113a..22e37f1ce5 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/service/FilesetMetaService.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/FilesetMetaService.java
@@ -466,15 +466,14 @@ public class FilesetMetaService {
* @param observedFilesetPO the fileset row and OCC version observed by the
caller
*/
void deleteFilesetWithVersion(NameIdentifier identifier, FilesetPO
observedFilesetPO) {
- int deleted =
- SessionUtils.getWithoutCommit(
- FilesetMetaMapper.class,
- mapper ->
- mapper.softDeleteFilesetMetasByFilesetId(
- observedFilesetPO.getFilesetId(),
observedFilesetPO.getCurrentVersion()));
- if (deleted == 0) {
- throw filesetWriteFailure(identifier, observedFilesetPO);
- }
+ OccWriteSupport.deleteWithVersion(
+ () ->
+ SessionUtils.getWithoutCommit(
+ FilesetMetaMapper.class,
+ mapper ->
+ mapper.softDeleteFilesetMetasByFilesetId(
+ observedFilesetPO.getFilesetId(),
observedFilesetPO.getCurrentVersion())),
+ () -> filesetWriteFailure(identifier, observedFilesetPO));
}
private boolean tryUpdateFileset(FilesetPO newFilesetPO, FilesetPO
oldFilesetPO) {
@@ -553,18 +552,16 @@ public class FilesetMetaService {
// The failed CAS has already serialized with an in-flight writer. A
non-locking natural-key
// lookup is enough to distinguish a disappeared name from one that still
names either the
// modified fileset or a replacement, without holding another row lock on
the failure path.
- Long currentFilesetId =
- SessionUtils.getWithoutCommit(
- FilesetMetaMapper.class,
- mapper ->
- mapper.selectFilesetIdBySchemaIdAndName(
- observedFilesetPO.getSchemaId(),
observedFilesetPO.getFilesetName()));
- if (currentFilesetId == null) {
- return new NoSuchEntityException(
- NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
- Entity.EntityType.FILESET.name().toLowerCase(),
- identifier.name());
- }
- return ExceptionUtils.concurrentModification(Entity.EntityType.FILESET,
identifier);
+ return OccWriteSupport.writeFailure(
+ identifier,
+ Entity.EntityType.FILESET,
+ () ->
+ SessionUtils.getWithoutCommit(
+ FilesetMetaMapper.class,
+ mapper ->
+ mapper.selectFilesetIdBySchemaIdAndName(
+ observedFilesetPO.getSchemaId(),
observedFilesetPO.getFilesetName())),
+ null,
+ null);
}
}
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 4902fa181c..b431cccf6f 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
@@ -246,17 +246,15 @@ public class GroupMetaService {
void deleteGroupWithVersion(NameIdentifier identifier, GroupPO
observedGroupPO) {
Long groupId = observedGroupPO.getGroupId();
SessionUtils.doMultipleWithCommit(
- () -> {
- int deleted =
- SessionUtils.getWithoutCommit(
- GroupMetaMapper.class,
- mapper ->
- mapper.softDeleteGroupMetaByGroupId(
- groupId, observedGroupPO.getCurrentVersion()));
- if (deleted == 0) {
- throw groupWriteFailure(identifier, observedGroupPO,
GroupLookup.NAME);
- }
- },
+ () ->
+ OccWriteSupport.deleteWithVersion(
+ () ->
+ SessionUtils.getWithoutCommit(
+ GroupMetaMapper.class,
+ mapper ->
+ mapper.softDeleteGroupMetaByGroupId(
+ groupId, observedGroupPO.getCurrentVersion())),
+ () -> groupWriteFailure(identifier, observedGroupPO,
GroupLookup.NAME)),
() ->
SessionUtils.doWithoutCommit(
GroupRoleRelMapper.class,
@@ -609,18 +607,16 @@ public class GroupMetaService {
* create for no reason.
*/
private void lockMetalakeForGroupCreate(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()));
}
private RuntimeException groupWriteFailure(
@@ -629,20 +625,15 @@ public class GroupMetaService {
// 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.
- GroupPO currentGroupPO =
getGroupPOByIdForUpdate(observedGroupPO.getGroupId());
- boolean missing =
- currentGroupPO == null
- || !Objects.equals(currentGroupPO.getMetalakeId(),
observedGroupPO.getMetalakeId());
- if (!missing && lookup == GroupLookup.NAME) {
- missing = !Objects.equals(currentGroupPO.getGroupName(),
observedGroupPO.getGroupName());
- }
- if (missing) {
- return new NoSuchEntityException(
- NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
- Entity.EntityType.GROUP.name().toLowerCase(),
- identifier.name());
- }
- return ExceptionUtils.concurrentModification(Entity.EntityType.GROUP,
identifier);
+ return OccWriteSupport.writeFailure(
+ identifier,
+ Entity.EntityType.GROUP,
+ () -> getGroupPOByIdForUpdate(observedGroupPO.getGroupId()),
+ null,
+ current ->
+ Objects.equals(current.getMetalakeId(),
observedGroupPO.getMetalakeId())
+ && (lookup != GroupLookup.NAME
+ || Objects.equals(current.getGroupName(),
observedGroupPO.getGroupName())));
}
private GroupPO getGroupPOByIdForUpdate(long groupId) {
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/ModelMetaService.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/ModelMetaService.java
index e49ad97328..a482cfb3ce 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/service/ModelMetaService.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/ModelMetaService.java
@@ -412,15 +412,14 @@ public class ModelMetaService {
* data. This allows any later cleanup failure to restore the model row as
well.
*/
void deleteModelWithVersion(NameIdentifier ident, ModelPO observedModelPO) {
- int deleted =
- SessionUtils.getWithoutCommit(
- ModelMetaMapper.class,
- mapper ->
- mapper.softDeleteModelMetaByIdAndVersion(
- observedModelPO.getModelId(),
observedModelPO.getCurrentVersion()));
- if (deleted == 0) {
- throw modelWriteFailure(ident, observedModelPO);
- }
+ OccWriteSupport.deleteWithVersion(
+ () ->
+ SessionUtils.getWithoutCommit(
+ ModelMetaMapper.class,
+ mapper ->
+ mapper.softDeleteModelMetaByIdAndVersion(
+ observedModelPO.getModelId(),
observedModelPO.getCurrentVersion())),
+ () -> modelWriteFailure(ident, observedModelPO));
}
/**
@@ -475,21 +474,20 @@ public class ModelMetaService {
* deleted, renamed, or moved, the requested model no longer exists.
*/
private RuntimeException modelWriteFailure(NameIdentifier ident, ModelPO
observedModelPO) {
- ModelPO currentModelPO =
- SessionUtils.getWithoutCommit(
- ModelMetaMapper.class,
- mapper ->
mapper.selectModelMetaByModelIdForUpdate(observedModelPO.getModelId()));
- if (currentModelPO == null
- || !Objects.equals(currentModelPO.getModelName(),
observedModelPO.getModelName())
- || !Objects.equals(currentModelPO.getSchemaId(),
observedModelPO.getSchemaId())
- || !Objects.equals(currentModelPO.getCatalogId(),
observedModelPO.getCatalogId())
- || !Objects.equals(currentModelPO.getMetalakeId(),
observedModelPO.getMetalakeId())) {
- return new NoSuchEntityException(
- NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
- Entity.EntityType.MODEL.name().toLowerCase(Locale.ROOT),
- ident.toString());
- }
- return ExceptionUtils.concurrentModification(Entity.EntityType.MODEL,
ident);
+ return OccWriteSupport.writeFailure(
+ ident,
+ Entity.EntityType.MODEL,
+ ident.toString(),
+ () ->
+ SessionUtils.getWithoutCommit(
+ ModelMetaMapper.class,
+ mapper ->
mapper.selectModelMetaByModelIdForUpdate(observedModelPO.getModelId())),
+ null,
+ current ->
+ Objects.equals(current.getModelName(),
observedModelPO.getModelName())
+ && Objects.equals(current.getSchemaId(),
observedModelPO.getSchemaId())
+ && Objects.equals(current.getCatalogId(),
observedModelPO.getCatalogId())
+ && Objects.equals(current.getMetalakeId(),
observedModelPO.getMetalakeId()));
}
private void deleteModelDependents(ModelPO modelPO) {
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
index 3e5004a896..f9797e0646 100644
---
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
@@ -42,8 +42,9 @@ public class 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
+ * <p>Executes a lookup to re-read the target entity. The caller is
responsible for choosing the
+ * lookup semantics needed by its transaction, usually a locking read by
stable ID. 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)}.
@@ -51,7 +52,7 @@ public class OccWriteSupport {
* @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 currentLookup a supplier that retrieves the current entity
* @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
@@ -59,10 +60,35 @@ public class OccWriteSupport {
public static <T> RuntimeException writeFailure(
NameIdentifier identifier,
Entity.EntityType type,
- Supplier<T> lockingLookup,
+ Supplier<T> currentLookup,
@Nullable Function<T, T> poMapper,
@Nullable Predicate<T> sameIdentity) {
- T currentPO = lockingLookup.get();
+ return writeFailure(identifier, type, identifier.name(), currentLookup,
poMapper, sameIdentity);
+ }
+
+ /**
+ * Classifies a write-failure while preserving a caller-specific entity name
in not-found errors.
+ *
+ * <p>This overload is useful for services whose existing error contract
reports a fully qualified
+ * name rather than {@link NameIdentifier#name()}.
+ *
+ * @param <T> the persistent object (PO) type of the entity
+ * @param identifier the name identifier used for an optimistic-lock error
+ * @param type the entity type
+ * @param entityName the entity name used for a not-found error
+ * @param currentLookup a supplier that retrieves the current entity
+ * @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,
+ String entityName,
+ Supplier<T> currentLookup,
+ @Nullable Function<T, T> poMapper,
+ @Nullable Predicate<T> sameIdentity) {
+ T currentPO = currentLookup.get();
if (currentPO != null && poMapper != null) {
currentPO = poMapper.apply(currentPO);
}
@@ -70,7 +96,7 @@ public class OccWriteSupport {
return new NoSuchEntityException(
NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
type.name().toLowerCase(Locale.ROOT),
- identifier.name());
+ entityName);
}
return ExceptionUtils.concurrentModification(type, identifier);
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/RoleMetaService.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/RoleMetaService.java
index 803d4949df..08d69202db 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/service/RoleMetaService.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/RoleMetaService.java
@@ -45,10 +45,12 @@ 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.GroupRoleRelMapper;
+import org.apache.gravitino.storage.relational.mapper.MetalakeMetaMapper;
import org.apache.gravitino.storage.relational.mapper.OwnerMetaMapper;
import org.apache.gravitino.storage.relational.mapper.RoleMetaMapper;
import org.apache.gravitino.storage.relational.mapper.SecurableObjectMapper;
import org.apache.gravitino.storage.relational.mapper.UserRoleRelMapper;
+import org.apache.gravitino.storage.relational.po.MetalakePO;
import org.apache.gravitino.storage.relational.po.RolePO;
import org.apache.gravitino.storage.relational.po.SecurableObjectPO;
import org.apache.gravitino.storage.relational.utils.ExceptionUtils;
@@ -154,8 +156,17 @@ public class RoleMetaService {
AuthorizationUtils.checkRole(roleEntity.nameIdentifier());
String metalake =
NameIdentifierUtil.getMetalake(roleEntity.nameIdentifier());
- Long metalakeId =
MetalakeMetaService.getInstance().getMetalakeIdByName(metalake);
- RolePO.Builder builder = RolePO.builder().withMetalakeId(metalakeId);
+ MetalakePO metalakePO =
+ SessionUtils.getWithoutCommit(
+ MetalakeMetaMapper.class, mapper ->
mapper.selectMetalakeMetaByName(metalake));
+ if (metalakePO == null) {
+ throw new NoSuchEntityException(
+ NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
+ Entity.EntityType.METALAKE.name().toLowerCase(),
+ metalake);
+ }
+
+ RolePO.Builder builder =
RolePO.builder().withMetalakeId(metalakePO.getMetalakeId());
RolePO rolePO = POConverters.initializeRolePOWithVersion(roleEntity,
builder);
List<SecurableObjectPO> securableObjectPOs = Lists.newArrayList();
for (SecurableObject object : roleEntity.securableObjects()) {
@@ -168,26 +179,30 @@ public class RoleMetaService {
securableObjectPOs.add(objectBuilder.build());
}
+ // The role row is written before its securable objects. Concurrent
overwrites then contend
+ // on the role row first and replace the child rows only after they are
serialized. The role
+ // and its securable objects therefore change atomically, with the last
overwrite winning.
SessionUtils.doMultipleWithCommit(
+ () -> lockMetalakeForRoleCreate(metalakePO),
() ->
SessionUtils.doWithoutCommit(
- SecurableObjectMapper.class,
+ RoleMetaMapper.class,
mapper -> {
if (overwritten) {
-
mapper.softDeleteSecurableObjectsByRoleId(rolePO.getRoleId());
- }
- if (!securableObjectPOs.isEmpty()) {
- mapper.batchInsertSecurableObjects(securableObjectPOs);
+ mapper.insertRoleMetaOnDuplicateKeyUpdate(rolePO);
+ } else {
+ mapper.insertRoleMeta(rolePO);
}
}),
() ->
SessionUtils.doWithoutCommit(
- RoleMetaMapper.class,
+ SecurableObjectMapper.class,
mapper -> {
if (overwritten) {
- mapper.insertRoleMetaOnDuplicateKeyUpdate(rolePO);
- } else {
- mapper.insertRoleMeta(rolePO);
+
mapper.softDeleteSecurableObjectsByRoleId(rolePO.getRoleId());
+ }
+ if (!securableObjectPOs.isEmpty()) {
+ mapper.batchInsertSecurableObjects(securableObjectPOs);
}
}));
@@ -225,10 +240,10 @@ public class RoleMetaService {
Set<SecurableObject> insertObjects = Sets.difference(newObjects,
oldObjects);
Set<SecurableObject> deleteObjects = Sets.difference(oldObjects,
newObjects);
- if (insertObjects.isEmpty() && deleteObjects.isEmpty()) {
- return newRoleEntity;
- }
-
+ // Every update runs the compare-and-set, including one that leaves the
securable objects
+ // untouched. The short-circuit that used to return early here would
skip the version check,
+ // so a caller whose snapshot was already stale would be told the update
succeeded. It also
+ // has to run because a metadata-only change, such as the audit info,
still has to be written.
List<SecurableObjectPO> deleteSecurableObjectPOs =
toSecurableObjectPOs(deleteObjects, oldRoleEntity, metalake);
@@ -236,12 +251,17 @@ public class RoleMetaService {
toSecurableObjectPOs(insertObjects, oldRoleEntity, metalake);
SessionUtils.doMultipleWithCommit(
- () ->
- SessionUtils.doWithoutCommit(
- RoleMetaMapper.class,
- mapper ->
- mapper.updateRoleMeta(
- POConverters.updateRolePOWithVersion(rolePO,
newRoleEntity), rolePO)),
+ () -> {
+ int updated =
+ SessionUtils.getWithoutCommit(
+ RoleMetaMapper.class,
+ mapper ->
+ mapper.updateRoleMeta(
+ POConverters.updateRolePOWithVersion(rolePO,
newRoleEntity), rolePO));
+ if (updated == 0) {
+ throw roleWriteFailure(identifier, rolePO);
+ }
+ },
() -> {
if (deleteSecurableObjectPOs.isEmpty()) {
return;
@@ -308,12 +328,32 @@ public class RoleMetaService {
Long metalakeId =
MetalakeMetaService.getInstance().getMetalakeIdByName(identifier.namespace().level(0));
- Long roleId = getRoleIdByMetalakeIdAndName(metalakeId, identifier.name());
+ RolePO rolePO = getRolePOByMetalakeIdAndName(metalakeId,
identifier.name());
+
+ deleteRoleWithVersion(identifier, rolePO);
+ return true;
+ }
+ /**
+ * Deletes the role whose version matches {@code observedRolePO}, together
with its role and owner
+ * relations. Package-private so tests can hand in a deliberately stale PO;
callers outside this
+ * class go through {@link #deleteRole(NameIdentifier)}, which reads the row
first.
+ *
+ * @param identifier the role being deleted, used only to build the error
+ * @param observedRolePO the role row the caller observed, carrying the
version to match
+ */
+ void deleteRoleWithVersion(NameIdentifier identifier, RolePO observedRolePO)
{
+ Long roleId = observedRolePO.getRoleId();
SessionUtils.doMultipleWithCommit(
() ->
- SessionUtils.doWithoutCommit(
- RoleMetaMapper.class, mapper ->
mapper.softDeleteRoleMetaByRoleId(roleId)),
+ OccWriteSupport.deleteWithVersion(
+ () ->
+ SessionUtils.getWithoutCommit(
+ RoleMetaMapper.class,
+ mapper ->
+ mapper.softDeleteRoleMetaByRoleId(
+ roleId, observedRolePO.getCurrentVersion())),
+ () -> roleWriteFailure(identifier, observedRolePO)),
() ->
SessionUtils.doWithoutCommit(
UserRoleRelMapper.class, mapper ->
mapper.softDeleteUserRoleRelByRoleId(roleId)),
@@ -330,7 +370,6 @@ public class RoleMetaService {
mapper ->
mapper.softDeleteOwnerRelByMetadataObjectIdAndType(
roleId, MetadataObject.Type.ROLE.name())));
- return true;
}
@Monitored(
@@ -455,6 +494,55 @@ public class RoleMetaService {
return rolePO;
}
+ /**
+ * Holds the parent metalake row for the rest of the transaction, so the
role cannot be created
+ * under a metalake that is going away.
+ *
+ * <p>The lock is shared, not exclusive: many roles can be created under the
same metalake at the
+ * same time. Dropping a metalake takes an exclusive lock on this row, so a
drop and a create
+ * cannot overlap. Whoever gets the row first wins, and the loser either
sees the metalake gone or
+ * inserts under a metalake that is still there.
+ *
+ * <p>The name is compared again because the ID alone cannot tell a rename
apart: the caller
+ * looked the metalake up by name, so a renamed row means the name in the
request no longer
+ * exists.
+ *
+ * <p>The metalake's version is deliberately not compared, matching {@code
CatalogMetaService}.
+ * Holding the row is what makes the create safe. An unrelated metalake edit
that commits in
+ * between bumps the version without making this create wrong, so comparing
it would reject the
+ * create for no reason.
+ */
+ private void lockMetalakeForRoleCreate(MetalakePO observedMetalakePO) {
+ OccWriteSupport.lockParentForChildWrite(
+ observedMetalakePO.getMetalakeName(),
+ Entity.EntityType.METALAKE,
+ () ->
+ SessionUtils.getWithoutCommit(
+ MetalakeMetaMapper.class,
+ mapper ->
+
mapper.selectMetalakeMetaByIdForShare(observedMetalakePO.getMetalakeId())),
+ null,
+ current -> Objects.equals(current.getMetalakeName(),
observedMetalakePO.getMetalakeName()));
+ }
+
+ private RuntimeException roleWriteFailure(NameIdentifier identifier, RolePO
observedRolePO) {
+ // 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.
+ return OccWriteSupport.writeFailure(
+ identifier,
+ Entity.EntityType.ROLE,
+ () ->
+ SessionUtils.getWithoutCommit(
+ RoleMetaMapper.class,
+ mapper ->
mapper.selectRoleMetaByIdForUpdate(observedRolePO.getRoleId())),
+ null,
+ current ->
+ Objects.equals(current.getRoleName(), observedRolePO.getRoleName())
+ && Objects.equals(current.getMetalakeId(),
observedRolePO.getMetalakeId()));
+ }
+
private static MetadataObject.Type getType(String type) {
return MetadataObject.Type.valueOf(type);
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/TopicMetaService.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/TopicMetaService.java
index 367c33de63..fbe9c87565 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/service/TopicMetaService.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/TopicMetaService.java
@@ -345,19 +345,17 @@ public class TopicMetaService {
*/
void deleteTopicWithVersion(NameIdentifier identifier, TopicPO
observedTopicPO) {
SessionUtils.doMultipleWithCommit(
- () -> {
- // Check the root version before touching relationships. If this
snapshot is stale,
- // throwing here stops the transaction before any dependent row can
be deleted.
- int deleted =
- SessionUtils.getWithoutCommit(
- TopicMetaMapper.class,
- mapper ->
- mapper.softDeleteTopicMetasByTopicId(
- observedTopicPO.getTopicId(),
observedTopicPO.getCurrentVersion()));
- if (deleted == 0) {
- throw topicWriteFailure(identifier, observedTopicPO);
- }
- },
+ // Check the root version before touching relationships. If this
snapshot is stale,
+ // throwing here stops the transaction before any dependent row can be
deleted.
+ () ->
+ OccWriteSupport.deleteWithVersion(
+ () ->
+ SessionUtils.getWithoutCommit(
+ TopicMetaMapper.class,
+ mapper ->
+ mapper.softDeleteTopicMetasByTopicId(
+ observedTopicPO.getTopicId(),
observedTopicPO.getCurrentVersion())),
+ () -> topicWriteFailure(identifier, observedTopicPO)),
() -> deleteTopicDependents(observedTopicPO.getTopicId()));
}
@@ -391,20 +389,18 @@ public class TopicMetaService {
// A zero-row CAS means either the same topic has a newer version, or the
topic disappeared
// from the name the caller used. The stable-ID lock waits for an
in-flight writer to finish so
// the result is classified from committed identity data.
- TopicPO currentTopicPO =
- SessionUtils.getWithoutCommit(
- TopicMetaMapper.class,
- mapper ->
mapper.selectTopicMetaByIdForUpdate(observedTopicPO.getTopicId()));
- if (currentTopicPO == null
- || !Objects.equals(currentTopicPO.getTopicName(),
observedTopicPO.getTopicName())
- || !Objects.equals(currentTopicPO.getSchemaId(),
observedTopicPO.getSchemaId())
- || !Objects.equals(currentTopicPO.getCatalogId(),
observedTopicPO.getCatalogId())
- || !Objects.equals(currentTopicPO.getMetalakeId(),
observedTopicPO.getMetalakeId())) {
- return new NoSuchEntityException(
- NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
- Entity.EntityType.TOPIC.name().toLowerCase(),
- identifier.name());
- }
- return ExceptionUtils.concurrentModification(Entity.EntityType.TOPIC,
identifier);
+ return OccWriteSupport.writeFailure(
+ identifier,
+ Entity.EntityType.TOPIC,
+ () ->
+ SessionUtils.getWithoutCommit(
+ TopicMetaMapper.class,
+ mapper ->
mapper.selectTopicMetaByIdForUpdate(observedTopicPO.getTopicId())),
+ null,
+ current ->
+ Objects.equals(current.getTopicName(),
observedTopicPO.getTopicName())
+ && Objects.equals(current.getSchemaId(),
observedTopicPO.getSchemaId())
+ && Objects.equals(current.getCatalogId(),
observedTopicPO.getCatalogId())
+ && Objects.equals(current.getMetalakeId(),
observedTopicPO.getMetalakeId()));
}
}
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 2ba8d38c9e..a4b088b4da 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
@@ -207,17 +207,15 @@ public class UserMetaService {
void deleteUserWithVersion(NameIdentifier identifier, UserPO observedUserPO)
{
Long userId = observedUserPO.getUserId();
SessionUtils.doMultipleWithCommit(
- () -> {
- int deleted =
- SessionUtils.getWithoutCommit(
- UserMetaMapper.class,
- mapper ->
- mapper.softDeleteUserMetaByUserId(
- userId, observedUserPO.getCurrentVersion()));
- if (deleted == 0) {
- throw userWriteFailure(identifier, observedUserPO,
UserLookup.NAME);
- }
- },
+ () ->
+ OccWriteSupport.deleteWithVersion(
+ () ->
+ SessionUtils.getWithoutCommit(
+ UserMetaMapper.class,
+ mapper ->
+ mapper.softDeleteUserMetaByUserId(
+ userId, observedUserPO.getCurrentVersion())),
+ () -> userWriteFailure(identifier, observedUserPO,
UserLookup.NAME)),
() ->
SessionUtils.doWithoutCommit(
UserRoleRelMapper.class, mapper ->
mapper.softDeleteUserRoleRelByUserId(userId)),
@@ -600,18 +598,16 @@ public class UserMetaService {
* create for no reason.
*/
private void lockMetalakeForUserCreate(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()));
}
private RuntimeException userWriteFailure(
@@ -620,22 +616,17 @@ public class UserMetaService {
// 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.
- UserPO currentUserPO = getUserPOByIdForUpdate(observedUserPO.getUserId());
- boolean missing =
- currentUserPO == null
- || !Objects.equals(currentUserPO.getMetalakeId(),
observedUserPO.getMetalakeId());
- if (!missing && lookup == UserLookup.NAME) {
- missing = !Objects.equals(currentUserPO.getUserName(),
observedUserPO.getUserName());
- } else if (!missing && lookup == UserLookup.EXTERNAL_ID) {
- missing = !Objects.equals(currentUserPO.getExternalId(),
observedUserPO.getExternalId());
- }
- if (missing) {
- return new NoSuchEntityException(
- NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
- Entity.EntityType.USER.name().toLowerCase(),
- identifier.name());
- }
- return ExceptionUtils.concurrentModification(Entity.EntityType.USER,
identifier);
+ return OccWriteSupport.writeFailure(
+ identifier,
+ Entity.EntityType.USER,
+ () -> getUserPOByIdForUpdate(observedUserPO.getUserId()),
+ null,
+ current ->
+ Objects.equals(current.getMetalakeId(),
observedUserPO.getMetalakeId())
+ && (lookup != UserLookup.NAME
+ || Objects.equals(current.getUserName(),
observedUserPO.getUserName()))
+ && (lookup != UserLookup.EXTERNAL_ID
+ || Objects.equals(current.getExternalId(),
observedUserPO.getExternalId())));
}
private UserPO getUserPOByIdForUpdate(long userId) {
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 45390dc926..1fec61c1d7 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
@@ -1367,11 +1367,15 @@ public class POConverters {
}
}
+ /**
+ * Updates a role PO and advances its OCC version.
+ *
+ * @param oldRolePO the role PO carrying the current version
+ * @param newRole the updated role entity
+ * @return a role PO whose current and last versions are advanced
+ */
public static RolePO updateRolePOWithVersion(RolePO oldRolePO, RoleEntity
newRole) {
- Long lastVersion = oldRolePO.getLastVersion();
- // TODO: set the version to the last version + 1 when having some fields
need be multiple
- // version
- Long nextVersion = lastVersion;
+ Long nextVersion = oldRolePO.getCurrentVersion() + 1;
try {
return RolePO.builder()
.withRoleId(oldRolePO.getRoleId())
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 86bfd5026c..39aa1e3f02 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
@@ -188,7 +188,7 @@ public class TestAuthMappers {
void testRoleMetaTouchUpdatedAtSkipsSoftDeleted() {
insertMetalake(1L, "metalake1");
insertRole(13L, "role13", 1L);
- roleMetaMapper.softDeleteRoleMetaByRoleId(13L);
+ roleMetaMapper.softDeleteRoleMetaByRoleId(13L, 1L);
long beforeUpdatedAt = queryUpdatedAt("role_meta", "role_id", 13L);
roleMetaMapper.touchRoleUpdatedAt(13L);
@@ -197,6 +197,16 @@ public class TestAuthMappers {
Assertions.assertEquals(beforeUpdatedAt, afterUpdatedAt);
}
+ @Test
+ void testRoleDeleteUsesCurrentVersion() {
+ insertMetalake(1L, "metalake1");
+ insertRole(14L, "role14", 1L);
+
+ Assertions.assertEquals(0, roleMetaMapper.softDeleteRoleMetaByRoleId(14L,
2L));
+
Assertions.assertNotNull(roleMetaMapper.selectRoleMetaByMetalakeIdAndName(1L,
"role14"));
+ Assertions.assertEquals(1, roleMetaMapper.softDeleteRoleMetaByRoleId(14L,
1L));
+ }
+
@Test
void testUserMetaTouchUpdatedAt() {
insertMetalake(1L, "metalake1");
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
index fb4a533b95..e4432034b8 100644
---
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
@@ -75,6 +75,16 @@ public class TestOccWriteSupport {
assertInstanceOf(NoSuchEntityException.class, ex);
}
+ @Test
+ void testWriteFailureUsesExplicitEntityName() {
+ NameIdentifier ident = NameIdentifier.of("metalake", "catalog", "schema",
"model");
+ RuntimeException ex =
+ OccWriteSupport.writeFailure(
+ ident, Entity.EntityType.MODEL, ident.toString(), () -> null,
null, po -> true);
+
+ assertEquals("No such model entity: metalake.catalog.schema.model",
ex.getMessage());
+ }
+
@Test
void testWriteFailureReturnsOptimisticLockExceptionWhenMatch() {
NameIdentifier ident = NameIdentifier.of("metalake_test");
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestRoleMetaService.java
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestRoleMetaService.java
index 287b346413..ecc034918c 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestRoleMetaService.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestRoleMetaService.java
@@ -46,6 +46,7 @@ import org.apache.gravitino.authorization.Privileges;
import org.apache.gravitino.authorization.SecurableObject;
import org.apache.gravitino.authorization.SecurableObjects;
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;
@@ -58,8 +59,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.GroupMetaMapper;
+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.GroupPO;
+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.session.SqlSessionFactoryHelper;
import org.apache.gravitino.storage.relational.utils.SessionUtils;
@@ -835,6 +840,196 @@ class TestRoleMetaService extends TestJDBCBackend {
Assertions.assertTrue(revokeMultipleRole.securableObjects().isEmpty());
}
+ @TestTemplate
+ void testConcurrentUpdateDoesNotChangeSecurableObjectsOnConflict() throws
IOException {
+ createAndInsertMakeLake(METALAKE_NAME);
+ createAndInsertCatalog(METALAKE_NAME, "catalog");
+ RoleEntity role =
+ createRoleEntity(
+ RandomIdGenerator.INSTANCE.nextId(),
+ AuthorizationUtils.ofRoleNamespace(METALAKE_NAME),
+ "concurrent-role",
+ AUDIT_INFO,
+ "catalog");
+ RoleMetaService.getInstance().insertRole(role, false);
+
+ Assertions.assertThrows(
+ OptimisticLockException.class,
+ () ->
+ RoleMetaService.getInstance()
+ .updateRole(
+ role.nameIdentifier(),
+ (RoleEntity oldRole) -> {
+ advanceRoleVersion(role.id());
+ List<SecurableObject> securableObjects =
+ Lists.newArrayList(oldRole.securableObjects());
+ securableObjects.add(
+ SecurableObjects.ofMetalake(
+ METALAKE_NAME,
Lists.newArrayList(Privileges.CreateTable.allow())));
+ return RoleEntity.builder()
+ .withId(oldRole.id())
+ .withName(oldRole.name())
+ .withNamespace(oldRole.namespace())
+ .withProperties(oldRole.properties())
+ .withSecurableObjects(securableObjects)
+ .withAuditInfo(oldRole.auditInfo())
+ .build();
+ }));
+
+ RoleEntity storedRole =
+
RoleMetaService.getInstance().getRoleByIdentifier(role.nameIdentifier());
+ assertTrue(
+ CollectionUtils.isEqualCollection(
+ Lists.newArrayList(
+ SecurableObjects.ofCatalog(
+ "catalog",
Lists.newArrayList(Privileges.UseCatalog.allow()))),
+ storedRole.securableObjects()));
+ }
+
+ @TestTemplate
+ void testCreateLocksMetalakeWithoutChangingVersion() throws IOException {
+ createAndInsertMakeLake(METALAKE_NAME);
+ createAndInsertCatalog(METALAKE_NAME, "catalog");
+ RoleMetaService service = RoleMetaService.getInstance();
+ MetalakePO beforeCreate = getMetalakePO();
+ RoleEntity role =
+ createRoleEntity(
+ RandomIdGenerator.INSTANCE.nextId(),
+ AuthorizationUtils.ofRoleNamespace(METALAKE_NAME),
+ "fenced-role",
+ AUDIT_INFO,
+ "catalog");
+
+ service.insertRole(role, false);
+
+ MetalakePO afterCreate = getMetalakePO();
+ assertEquals(beforeCreate.getCurrentVersion(),
afterCreate.getCurrentVersion());
+ assertEquals(beforeCreate.getLastVersion(), afterCreate.getLastVersion());
+
+ RoleEntity duplicate =
+ createRoleEntity(
+ RandomIdGenerator.INSTANCE.nextId(),
+ AuthorizationUtils.ofRoleNamespace(METALAKE_NAME),
+ role.name(),
+ AUDIT_INFO,
+ "catalog");
+ Assertions.assertThrows(
+ EntityAlreadyExistsException.class, () ->
service.insertRole(duplicate, false));
+
+ MetalakePO afterFailedCreate = getMetalakePO();
+ assertEquals(afterCreate.getCurrentVersion(),
afterFailedCreate.getCurrentVersion());
+ assertEquals(afterCreate.getLastVersion(),
afterFailedCreate.getLastVersion());
+ }
+
+ @TestTemplate
+ void testOverwriteInsertAdvancesVersion() throws IOException {
+ createAndInsertMakeLake(METALAKE_NAME);
+ createAndInsertCatalog(METALAKE_NAME, "catalog");
+ RoleMetaService service = RoleMetaService.getInstance();
+ RoleEntity role =
+ createRoleEntity(
+ RandomIdGenerator.INSTANCE.nextId(),
+ AuthorizationUtils.ofRoleNamespace(METALAKE_NAME),
+ "overwrite-role",
+ AUDIT_INFO,
+ "catalog");
+ service.insertRole(role, false);
+ RolePO initialPO = getRolePO(role.name());
+
+ service.insertRole(role, true);
+
+ RolePO overwrittenPO = getRolePO(role.name());
+ assertEquals(initialPO.getCurrentVersion() + 1,
overwrittenPO.getCurrentVersion());
+ assertEquals(overwrittenPO.getCurrentVersion(),
overwrittenPO.getLastVersion());
+ int staleDelete =
+ SessionUtils.doWithCommitAndFetchResult(
+ RoleMetaMapper.class,
+ mapper -> mapper.softDeleteRoleMetaByRoleId(role.id(),
initialPO.getCurrentVersion()));
+ assertEquals(0, staleDelete);
+ }
+
+ @TestTemplate
+ void testMetadataOnlyUpdateUsesOcc() throws IOException {
+ createAndInsertMakeLake(METALAKE_NAME);
+ createAndInsertCatalog(METALAKE_NAME, "catalog");
+ RoleMetaService service = RoleMetaService.getInstance();
+ RoleEntity role =
+ createRoleEntity(
+ RandomIdGenerator.INSTANCE.nextId(),
+ AuthorizationUtils.ofRoleNamespace(METALAKE_NAME),
+ "metadata-only-role",
+ AUDIT_INFO,
+ "catalog");
+ service.insertRole(role, false);
+ RolePO beforeUpdate = getRolePO(role.name());
+
+ service.updateRole(role.nameIdentifier(), (RoleEntity oldRole) ->
copyRole(oldRole, "value-1"));
+
+ RolePO afterUpdate = getRolePO(role.name());
+ assertEquals(beforeUpdate.getCurrentVersion() + 1,
afterUpdate.getCurrentVersion());
+ assertEquals(
+ "value-1",
service.getRoleByIdentifier(role.nameIdentifier()).properties().get("key"));
+
+ Assertions.assertThrows(
+ OptimisticLockException.class,
+ () ->
+ service.updateRole(
+ role.nameIdentifier(),
+ (RoleEntity oldRole) -> {
+ advanceRoleVersion(role.id());
+ return copyRole(oldRole, "value-2");
+ }));
+ assertEquals(
+ "value-1",
service.getRoleByIdentifier(role.nameIdentifier()).properties().get("key"));
+ }
+
+ @TestTemplate
+ void testStaleDeleteReportsConflict() throws IOException {
+ createAndInsertMakeLake(METALAKE_NAME);
+ createAndInsertCatalog(METALAKE_NAME, "catalog");
+ RoleMetaService service = RoleMetaService.getInstance();
+ RoleEntity role =
+ createRoleEntity(
+ RandomIdGenerator.INSTANCE.nextId(),
+ AuthorizationUtils.ofRoleNamespace(METALAKE_NAME),
+ "stale-delete-role",
+ AUDIT_INFO,
+ "catalog");
+ service.insertRole(role, false);
+ RolePO staleRolePO = getRolePO(role.name());
+ advanceRoleVersion(role.id());
+
+ Assertions.assertThrows(
+ OptimisticLockException.class,
+ () -> service.deleteRoleWithVersion(role.nameIdentifier(),
staleRolePO));
+ assertEquals(role.id(),
service.getRoleByIdentifier(role.nameIdentifier()).id());
+ }
+
+ @TestTemplate
+ void testAlterReportsNoSuchWhenRoleIsDeletedConcurrently() throws
IOException {
+ createAndInsertMakeLake(METALAKE_NAME);
+ createAndInsertCatalog(METALAKE_NAME, "catalog");
+ RoleMetaService service = RoleMetaService.getInstance();
+ RoleEntity role =
+ createRoleEntity(
+ RandomIdGenerator.INSTANCE.nextId(),
+ AuthorizationUtils.ofRoleNamespace(METALAKE_NAME),
+ "deleted-during-alter",
+ AUDIT_INFO,
+ "catalog");
+ service.insertRole(role, false);
+
+ Assertions.assertThrows(
+ NoSuchEntityException.class,
+ () ->
+ service.updateRole(
+ role.nameIdentifier(),
+ (RoleEntity oldRole) -> {
+ service.deleteRole(role.nameIdentifier());
+ return copyRole(oldRole, "ignored-value");
+ }));
+ }
+
@TestTemplate
void testDeleteMetalakeCascade() throws IOException {
BaseMetalake metalake = createAndInsertMakeLake(METALAKE_NAME);
@@ -1059,6 +1254,29 @@ class TestRoleMetaService extends TestJDBCBackend {
return count;
}
+ private MetalakePO getMetalakePO() {
+ return SessionUtils.getWithoutCommit(
+ MetalakeMetaMapper.class, mapper ->
mapper.selectMetalakeMetaByName(METALAKE_NAME));
+ }
+
+ private RolePO getRolePO(String roleName) {
+ MetalakePO metalakePO = getMetalakePO();
+ return SessionUtils.getWithoutCommit(
+ RoleMetaMapper.class,
+ mapper ->
mapper.selectRoleMetaByMetalakeIdAndName(metalakePO.getMetalakeId(), roleName));
+ }
+
+ private RoleEntity copyRole(RoleEntity role, String propertyValue) {
+ return RoleEntity.builder()
+ .withId(role.id())
+ .withName(role.name())
+ .withNamespace(role.namespace())
+ .withProperties(ImmutableMap.of("key", propertyValue))
+ .withSecurableObjects(role.securableObjects())
+ .withAuditInfo(role.auditInfo())
+ .build();
+ }
+
private Integer countUserRoleRels() {
int count = 0;
try (SqlSession sqlSession =
@@ -1127,4 +1345,19 @@ class TestRoleMetaService extends TestJDBCBackend {
}
return count;
}
+
+ private void advanceRoleVersion(long roleId) {
+ try (SqlSession sqlSession =
+
SqlSessionFactoryHelper.getInstance().getSqlSessionFactory().openSession(true);
+ Connection connection = sqlSession.getConnection();
+ Statement statement = connection.createStatement()) {
+ assertEquals(
+ 1,
+ statement.executeUpdate(
+ "UPDATE role_meta SET current_version = current_version + 1
WHERE role_id = "
+ + roleId));
+ } catch (SQLException e) {
+ throw new RuntimeException("Advance role 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 afb28a80f9..bce2ac19d8 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
@@ -55,6 +55,7 @@ import org.apache.gravitino.meta.GroupEntity;
import org.apache.gravitino.meta.ModelEntity;
import org.apache.gravitino.meta.ModelVersionEntity;
import org.apache.gravitino.meta.PolicyEntity;
+import org.apache.gravitino.meta.RoleEntity;
import org.apache.gravitino.meta.SchemaEntity;
import org.apache.gravitino.meta.SchemaVersion;
import org.apache.gravitino.meta.StatisticEntity;
@@ -89,6 +90,7 @@ import
org.apache.gravitino.storage.relational.po.ModelVersionPO;
import org.apache.gravitino.storage.relational.po.OwnerRelPO;
import org.apache.gravitino.storage.relational.po.PolicyPO;
import org.apache.gravitino.storage.relational.po.PolicyVersionPO;
+import org.apache.gravitino.storage.relational.po.RolePO;
import org.apache.gravitino.storage.relational.po.SchemaPO;
import org.apache.gravitino.storage.relational.po.SecurableObjectPO;
import org.apache.gravitino.storage.relational.po.StatisticPO;
@@ -765,6 +767,32 @@ public class TestPOConverters {
assertEquals("test", updatePO.getTableName());
}
+ @Test
+ public void testUpdateRolePOVersionUsesCurrentVersion() {
+ AuditInfo auditInfo =
+
AuditInfo.builder().withCreator("creator").withCreateTime(FIX_INSTANT).build();
+ RoleEntity role =
+
RoleEntity.builder().withId(1L).withName("role").withAuditInfo(auditInfo).build();
+ RolePO initialRolePO =
+ POConverters.initializeRolePOWithVersion(role,
RolePO.builder().withMetalakeId(1L));
+ RolePO rolePO =
+ RolePO.builder()
+ .withRoleId(initialRolePO.getRoleId())
+ .withRoleName(initialRolePO.getRoleName())
+ .withMetalakeId(initialRolePO.getMetalakeId())
+ .withProperties(initialRolePO.getProperties())
+ .withAuditInfo(initialRolePO.getAuditInfo())
+ .withCurrentVersion(7L)
+ .withLastVersion(3L)
+ .withDeletedAt(initialRolePO.getDeletedAt())
+ .build();
+
+ RolePO updatedRolePO = POConverters.updateRolePOWithVersion(rolePO, role);
+
+ assertEquals(8, updatedRolePO.getCurrentVersion());
+ assertEquals(8, updatedRolePO.getLastVersion());
+ }
+
@Test
public void testUpdateFilesetPOVersion() throws JsonProcessingException {
Map<String, String> properties = new HashMap<>();
