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 1b2b9c0735 [#12738] improvement(core): add OCC for group writes
(#12745)
1b2b9c0735 is described below
commit 1b2b9c0735d35e7da5b8c8642ae698d83eef894e
Author: Qi Yu <[email protected]>
AuthorDate: Tue Sep 1 11:00:25 2026 +0800
[#12738] improvement(core): add OCC for group writes (#12745)
### What changes were proposed in this pull request?
Add version-CAS optimistic concurrency control and atomic service
operations for managed groups.
- Increment the group version on every update and match updates by group
ID, expected version, and active-row state.
- Make soft deletes version checked and classify a zero-row result as
either a missing group or an OCC conflict.
- Execute the group CAS before group-role and ownership mutations in one
transaction.
- Fence the parent metalake while creating a group and advance the
version on overwrite upserts.
- Add coverage for metadata-only and ID updates, stale deletes,
overwrite versioning, parent fencing, and relationship rollback.
### Why are the changes needed?
Concurrent group mutations could otherwise overwrite each other or leave
partially updated relationship state.
Fix: #12738
### Does this PR introduce _any_ user-facing change?
Concurrent managed-group 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:test --tests TestGroupMetaService --tests
TestAuthMappers --tests TestPOConverters -PskipITs
-PskipDockerTests=true`
- `./gradlew :core:spotlessCheck :core:compileTestJava`
- Added a converter regression where `currentVersion` and `lastVersion`
differ.
- GitHub Backend Integration Test matrix: H2, MySQL, and PostgreSQL.
---
.../storage/relational/mapper/GroupMetaMapper.java | 12 +-
.../mapper/GroupMetaSQLProviderFactory.java | 10 +-
.../provider/base/GroupMetaBaseSQLProvider.java | 27 ++-
.../postgresql/GroupMetaPostgreSQLProvider.java | 14 +-
.../relational/service/GroupMetaService.java | 224 ++++++++++++++----
.../storage/relational/utils/POConverters.java | 5 +-
.../mapper/provider/base/TestAuthMappers.java | 12 +-
.../relational/service/TestGroupMetaService.java | 250 ++++++++++++++++++++-
.../storage/relational/utils/TestPOConverters.java | 28 +++
9 files changed, 507 insertions(+), 75 deletions(-)
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/GroupMetaMapper.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/GroupMetaMapper.java
index 8c30152d25..16bec4d7d2 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/GroupMetaMapper.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/GroupMetaMapper.java
@@ -53,6 +53,10 @@ public interface GroupMetaMapper {
GroupPO selectGroupMetaByMetalakeIdAndName(
@Param("metalakeId") Long metalakeId, @Param("groupName") String name);
+ /** Returns and locks an active group by ID for the current transaction. */
+ @SelectProvider(type = GroupMetaSQLProviderFactory.class, method =
"selectGroupMetaByIdForUpdate")
+ GroupPO selectGroupMetaByIdForUpdate(@Param("groupId") Long groupId);
+
@SelectProvider(
type = GroupMetaSQLProviderFactory.class,
method = "listExtendedGroupPOsByMetalakeIdAndNames")
@@ -88,8 +92,14 @@ public interface GroupMetaMapper {
method = "insertGroupMetaOnDuplicateKeyUpdate")
void insertGroupMetaOnDuplicateKeyUpdate(@Param("groupMeta") GroupPO
groupPO);
+ /**
+ * Soft-deletes an active group only when its OCC version still matches.
+ *
+ * @return the number of deleted rows
+ */
@UpdateProvider(type = GroupMetaSQLProviderFactory.class, method =
"softDeleteGroupMetaByGroupId")
- void softDeleteGroupMetaByGroupId(@Param("groupId") Long groupId);
+ Integer softDeleteGroupMetaByGroupId(
+ @Param("groupId") Long groupId, @Param("currentVersion") Long
currentVersion);
@UpdateProvider(
type = GroupMetaSQLProviderFactory.class,
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/GroupMetaSQLProviderFactory.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/GroupMetaSQLProviderFactory.java
index 9fa4eece50..a938a5e5ed 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/GroupMetaSQLProviderFactory.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/GroupMetaSQLProviderFactory.java
@@ -59,6 +59,11 @@ public class GroupMetaSQLProviderFactory {
return getProvider().selectGroupMetaByMetalakeIdAndName(metalakeId, name);
}
+ /** Returns SQL that selects and locks an active group by ID. */
+ public static String selectGroupMetaByIdForUpdate(@Param("groupId") Long
groupId) {
+ return getProvider().selectGroupMetaByIdForUpdate(groupId);
+ }
+
public static String listExtendedGroupPOsByMetalakeIdAndNames(
@Param("metalakeId") Long metalakeId, @Param("groupNames") List<String>
groupNames) {
return getProvider().listExtendedGroupPOsByMetalakeIdAndNames(metalakeId,
groupNames);
@@ -72,8 +77,9 @@ public class GroupMetaSQLProviderFactory {
return getProvider().insertGroupMetaOnDuplicateKeyUpdate(groupPO);
}
- public static String softDeleteGroupMetaByGroupId(@Param("groupId") Long
groupId) {
- return getProvider().softDeleteGroupMetaByGroupId(groupId);
+ public static String softDeleteGroupMetaByGroupId(
+ @Param("groupId") Long groupId, @Param("currentVersion") Long
currentVersion) {
+ return getProvider().softDeleteGroupMetaByGroupId(groupId, currentVersion);
}
public static String softDeleteGroupMetasByMetalakeId(@Param("metalakeId")
Long metalakeId) {
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/GroupMetaBaseSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/GroupMetaBaseSQLProvider.java
index 852e48cc42..cbc1a6a8d3 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/GroupMetaBaseSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/GroupMetaBaseSQLProvider.java
@@ -138,6 +138,17 @@ public class GroupMetaBaseSQLProvider {
+ " AND deleted_at = 0";
}
+ /** Returns SQL that selects and locks an active group by ID. */
+ public String selectGroupMetaByIdForUpdate(@Param("groupId") Long groupId) {
+ return "SELECT group_id as groupId, group_name as groupName,"
+ + " metalake_id as metalakeId, external_id as externalId, audit_info
as auditInfo,"
+ + " current_version as currentVersion, last_version as lastVersion,"
+ + " deleted_at as deletedAt"
+ + " FROM "
+ + GROUP_TABLE_NAME
+ + " WHERE group_id = #{groupId} AND deleted_at = 0 FOR UPDATE";
+ }
+
public String selectGroupMetaByMetalakeNameAndExternalId(
@Param("metalakeName") String metalakeName, @Param("externalId") String
externalId) {
return "SELECT gt.group_id as groupId, gt.group_name as groupName,"
@@ -245,17 +256,21 @@ public class GroupMetaBaseSQLProvider {
+ " metalake_id = #{groupMeta.metalakeId},"
+ " audit_info = #{groupMeta.auditInfo},"
+ " external_id = #{groupMeta.externalId},"
- + " current_version = #{groupMeta.currentVersion},"
- + " last_version = #{groupMeta.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 = #{groupMeta.deletedAt}";
}
- public String softDeleteGroupMetaByGroupId(@Param("groupId") Long groupId) {
+ public String softDeleteGroupMetaByGroupId(
+ @Param("groupId") Long groupId, @Param("currentVersion") Long
currentVersion) {
return "UPDATE "
+ GROUP_TABLE_NAME
+ " SET deleted_at = (UNIX_TIMESTAMP() * 1000.0)"
+ " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000"
- + " WHERE group_id = #{groupId} AND deleted_at = 0";
+ + " WHERE group_id = #{groupId}"
+ + " AND current_version = #{currentVersion} AND deleted_at = 0";
}
public String softDeleteGroupMetasByMetalakeId(@Param("metalakeId") Long
metalakeId) {
@@ -278,11 +293,7 @@ public class GroupMetaBaseSQLProvider {
+ " last_version = #{newGroupMeta.lastVersion},"
+ " deleted_at = #{newGroupMeta.deletedAt}"
+ " WHERE group_id = #{oldGroupMeta.groupId}"
- + " AND group_name = #{oldGroupMeta.groupName}"
- + " AND metalake_id = #{oldGroupMeta.metalakeId}"
- + " AND audit_info = #{oldGroupMeta.auditInfo}"
+ " AND current_version = #{oldGroupMeta.currentVersion}"
- + " AND last_version = #{oldGroupMeta.lastVersion}"
+ " AND deleted_at = 0";
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/GroupMetaPostgreSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/GroupMetaPostgreSQLProvider.java
index 450e26cef2..a5cef46358 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/GroupMetaPostgreSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/GroupMetaPostgreSQLProvider.java
@@ -30,11 +30,12 @@ import org.apache.ibatis.annotations.Param;
public class GroupMetaPostgreSQLProvider extends GroupMetaBaseSQLProvider {
@Override
- public String softDeleteGroupMetaByGroupId(Long groupId) {
+ public String softDeleteGroupMetaByGroupId(Long groupId, Long
currentVersion) {
return "UPDATE "
+ GROUP_TABLE_NAME
+ " SET deleted_at = CAST(EXTRACT(EPOCH FROM CURRENT_TIMESTAMP) * 1000
AS BIGINT)"
- + " WHERE group_id = #{groupId} AND deleted_at = 0";
+ + " WHERE group_id = #{groupId}"
+ + " AND current_version = #{currentVersion} AND deleted_at = 0";
}
@Override
@@ -67,8 +68,13 @@ public class GroupMetaPostgreSQLProvider extends
GroupMetaBaseSQLProvider {
+ " metalake_id = #{groupMeta.metalakeId},"
+ " external_id = #{groupMeta.externalId},"
+ " audit_info = #{groupMeta.auditInfo},"
- + " current_version = #{groupMeta.currentVersion},"
- + " last_version = #{groupMeta.lastVersion},"
+ // PostgreSQL requires the stored-row column to be qualified in ON
CONFLICT assignments.
+ + " current_version = "
+ + GROUP_TABLE_NAME
+ + ".current_version + 1,"
+ + " last_version = "
+ + GROUP_TABLE_NAME
+ + ".current_version + 1,"
+ " deleted_at = #{groupMeta.deletedAt}";
}
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 1229fc0c49..4902fa181c 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
@@ -29,6 +29,7 @@ import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Function;
import java.util.stream.Collectors;
import org.apache.gravitino.Entity;
@@ -43,10 +44,12 @@ import org.apache.gravitino.meta.RoleEntity;
import org.apache.gravitino.metrics.Monitored;
import org.apache.gravitino.storage.relational.mapper.GroupMetaMapper;
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.po.ExtendedGroupPO;
import org.apache.gravitino.storage.relational.po.GroupPO;
import org.apache.gravitino.storage.relational.po.GroupRoleRelPO;
+import org.apache.gravitino.storage.relational.po.MetalakePO;
import org.apache.gravitino.storage.relational.po.RolePO;
import org.apache.gravitino.storage.relational.utils.ExceptionUtils;
import org.apache.gravitino.storage.relational.utils.POConverters;
@@ -172,24 +175,34 @@ public class GroupMetaService {
NameIdentifier metalakeIdent =
NameIdentifier.of(NameIdentifierUtil.getMetalake(groupEntity.nameIdentifier()));
- Long metalakeId = EntityIdService.getEntityId(metalakeIdent,
Entity.EntityType.METALAKE);
-
- GroupPO.Builder builder = GroupPO.builder().withMetalakeId(metalakeId);
- GroupPO GroupPO = POConverters.initializeGroupPOWithVersion(groupEntity,
builder);
+ MetalakePO metalakePO =
+ SessionUtils.getWithoutCommit(
+ MetalakeMetaMapper.class,
+ mapper -> mapper.selectMetalakeMetaByName(metalakeIdent.name()));
+ if (metalakePO == null) {
+ throw new NoSuchEntityException(
+ NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
+ Entity.EntityType.METALAKE.name().toLowerCase(),
+ metalakeIdent.name());
+ }
+
+ GroupPO.Builder builder =
GroupPO.builder().withMetalakeId(metalakePO.getMetalakeId());
+ GroupPO groupPO = POConverters.initializeGroupPOWithVersion(groupEntity,
builder);
List<Long> roleIds =
Optional.ofNullable(groupEntity.roleIds()).orElse(Lists.newArrayList());
List<GroupRoleRelPO> groupRoleRelPOS =
POConverters.initializeGroupRoleRelsPOWithVersion(groupEntity,
roleIds);
SessionUtils.doMultipleWithCommit(
+ () -> lockMetalakeForGroupCreate(metalakePO),
() ->
SessionUtils.doWithoutCommit(
GroupMetaMapper.class,
mapper -> {
if (overwritten) {
- mapper.insertGroupMetaOnDuplicateKeyUpdate(GroupPO);
+ mapper.insertGroupMetaOnDuplicateKeyUpdate(groupPO);
} else {
- mapper.insertGroupMeta(GroupPO);
+ mapper.insertGroupMeta(groupPO);
}
}),
() -> {
@@ -215,12 +228,35 @@ public class GroupMetaService {
public boolean deleteGroup(NameIdentifier identifier) {
AuthorizationUtils.checkGroup(identifier);
- Long groupId = EntityIdService.getEntityId(identifier,
Entity.EntityType.GROUP);
+ Long metalakeId =
+
MetalakeMetaService.getInstance().getMetalakeIdByName(identifier.namespace().level(0));
+ GroupPO groupPO = getGroupPOByMetalakeIdAndName(metalakeId,
identifier.name());
+ deleteGroupWithVersion(identifier, groupPO);
+ return true;
+ }
+ /**
+ * Deletes the group whose version matches {@code observedGroupPO}, 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 #deleteGroup(NameIdentifier)}, which reads
the row first.
+ *
+ * @param identifier the group being deleted, used only to build the error
+ * @param observedGroupPO the group row the caller observed, carrying the
version to match
+ */
+ void deleteGroupWithVersion(NameIdentifier identifier, GroupPO
observedGroupPO) {
+ Long groupId = observedGroupPO.getGroupId();
SessionUtils.doMultipleWithCommit(
- () ->
- SessionUtils.doWithoutCommit(
- GroupMetaMapper.class, mapper ->
mapper.softDeleteGroupMetaByGroupId(groupId)),
+ () -> {
+ int deleted =
+ SessionUtils.getWithoutCommit(
+ GroupMetaMapper.class,
+ mapper ->
+ mapper.softDeleteGroupMetaByGroupId(
+ groupId, observedGroupPO.getCurrentVersion()));
+ if (deleted == 0) {
+ throw groupWriteFailure(identifier, observedGroupPO,
GroupLookup.NAME);
+ }
+ },
() ->
SessionUtils.doWithoutCommit(
GroupRoleRelMapper.class,
@@ -231,7 +267,6 @@ public class GroupMetaService {
mapper ->
mapper.softDeleteOwnerRelByOwnerIdAndType(
groupId, Entity.EntityType.GROUP.name())));
- return true;
}
@Monitored(metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
baseMetricName = "updateGroup")
@@ -265,18 +300,24 @@ public class GroupMetaService {
Set<Long> insertRoleIds = Sets.difference(newRoleIds, oldRoleIds);
Set<Long> deleteRoleIds = Sets.difference(oldRoleIds, newRoleIds);
- if (insertRoleIds.isEmpty() && deleteRoleIds.isEmpty()) {
- return newEntity;
- }
+ // Every update runs the compare-and-set, including one that leaves the
roles 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.
try {
SessionUtils.doMultipleWithCommit(
- () ->
- SessionUtils.doWithoutCommit(
- GroupMetaMapper.class,
- mapper ->
- mapper.updateGroupMeta(
- POConverters.updateGroupPOWithVersion(oldGroupPO,
newEntity),
- oldGroupPO)),
+ () -> {
+ int updated =
+ SessionUtils.getWithoutCommit(
+ GroupMetaMapper.class,
+ mapper ->
+ mapper.updateGroupMeta(
+ POConverters.updateGroupPOWithVersion(oldGroupPO,
newEntity),
+ oldGroupPO));
+ if (updated == 0) {
+ throw groupWriteFailure(identifier, oldGroupPO,
GroupLookup.NAME);
+ }
+ },
() -> {
if (insertRoleIds.isEmpty()) {
return;
@@ -440,13 +481,20 @@ public class GroupMetaService {
try {
SessionUtils.doMultipleWithCommit(
- () ->
- SessionUtils.doWithoutCommit(
- GroupMetaMapper.class,
- mapper ->
- mapper.updateGroupMeta(
- POConverters.updateGroupPOWithVersion(oldGroupPO,
newEntity),
- oldGroupPO)),
+ () -> {
+ int updated =
+ SessionUtils.getWithoutCommit(
+ GroupMetaMapper.class,
+ mapper ->
+ mapper.updateGroupMeta(
+ POConverters.updateGroupPOWithVersion(oldGroupPO,
newEntity),
+ oldGroupPO));
+ if (updated == 0) {
+ NameIdentifier groupIdIdentifier =
+ AuthorizationUtils.ofGroup(metalake,
String.valueOf(groupId));
+ throw groupWriteFailure(groupIdIdentifier, oldGroupPO,
GroupLookup.ID);
+ }
+ },
() ->
SessionUtils.doWithoutCommit(
GroupMetaMapper.class,
@@ -463,27 +511,45 @@ public class GroupMetaService {
metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
baseMetricName = "deleteGroupById")
public boolean deleteGroupById(String metalake, long groupId) {
+ GroupPO groupPO;
try {
- getGroupPOByMetalakeNameAndId(metalake, groupId);
+ groupPO = getGroupPOByMetalakeNameAndId(metalake, groupId);
} catch (NoSuchEntityException e) {
return false;
}
+ NameIdentifier identifier = AuthorizationUtils.ofGroup(metalake,
groupPO.getGroupName());
+
+ // Starts false so that any path that does not reach the child cleanup
reports "nothing was
+ // deleted here" rather than claiming a delete it did not perform.
+ AtomicBoolean deletedGroup = new AtomicBoolean(false);
SessionUtils.doMultipleWithCommit(
- () ->
- SessionUtils.doWithoutCommit(
- GroupMetaMapper.class, mapper ->
mapper.softDeleteGroupMetaByGroupId(groupId)),
- () ->
- SessionUtils.doWithoutCommit(
- GroupRoleRelMapper.class,
- mapper -> mapper.softDeleteGroupRoleRelByGroupId(groupId)),
- () ->
- SessionUtils.doWithoutCommit(
- OwnerMetaMapper.class,
- mapper ->
- mapper.softDeleteOwnerRelByOwnerIdAndType(
- groupId, Entity.EntityType.GROUP.name())));
- return true;
+ () -> {
+ int deleted =
+ SessionUtils.getWithoutCommit(
+ GroupMetaMapper.class,
+ mapper ->
+ mapper.softDeleteGroupMetaByGroupId(groupId,
groupPO.getCurrentVersion()));
+ if (deleted == 0) {
+ // The compare-and-set matched no row for one of two reasons.
Either the row is already
+ // gone, and a delete that has nothing left to delete is a no-op
rather than an error,
+ // or the row is still there under a newer version, which is a
genuine conflict.
+ if (getGroupPOByIdForUpdate(groupId) == null) {
+ return;
+ }
+ throw
ExceptionUtils.concurrentModification(Entity.EntityType.GROUP, identifier);
+ }
+
+ deletedGroup.set(true);
+ SessionUtils.doWithoutCommit(
+ GroupRoleRelMapper.class, mapper ->
mapper.softDeleteGroupRoleRelByGroupId(groupId));
+ SessionUtils.doWithoutCommit(
+ OwnerMetaMapper.class,
+ mapper ->
+ mapper.softDeleteOwnerRelByOwnerIdAndType(
+ groupId, Entity.EntityType.GROUP.name()));
+ });
+ return deletedGroup.get();
}
@Monitored(
@@ -523,4 +589,76 @@ public class GroupMetaService {
.collect(Collectors.toList());
return new PagedResult<>(totalCount, groups);
}
+
+ /**
+ * Holds the parent metalake row for the rest of the transaction, so the
group cannot be created
+ * under a metalake that is going away.
+ *
+ * <p>The lock is shared, not exclusive: many groups 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 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());
+ }
+ }
+
+ private RuntimeException groupWriteFailure(
+ NameIdentifier identifier, GroupPO observedGroupPO, GroupLookup lookup) {
+ // 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.
+ 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);
+ }
+
+ private GroupPO getGroupPOByIdForUpdate(long groupId) {
+ return SessionUtils.getWithoutCommit(
+ GroupMetaMapper.class, mapper ->
mapper.selectGroupMetaByIdForUpdate(groupId));
+ }
+
+ /**
+ * How the caller addressed the group, which decides what counts as "the
same group" when a failed
+ * compare-and-set is classified. A caller that used the name is looking for
that name, so a
+ * rename means the group it asked for is gone. A caller that used the ID
addressed the row
+ * itself, so a rename leaves it addressing the same group and only the
metalake has to still
+ * match.
+ */
+ private enum GroupLookup {
+ NAME,
+ ID
+ }
}
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 c05cd50d9b..45390dc926 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
@@ -1243,10 +1243,7 @@ public class POConverters {
* @return GroupPO object with updated version
*/
public static GroupPO updateGroupPOWithVersion(GroupPO oldGroupPO,
GroupEntity newGroup) {
- Long lastVersion = oldGroupPO.getLastVersion();
- // TODO: set the version to the last version + 1 when having some fields
need be multiple
- // version
- Long nextVersion = lastVersion;
+ Long nextVersion = oldGroupPO.getCurrentVersion() + 1;
try {
return GroupPO.builder()
.withGroupId(oldGroupPO.getGroupId())
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 550ad3c54c..86bfd5026c 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
@@ -264,7 +264,7 @@ public class TestAuthMappers {
void testGroupMetaTouchUpdatedAtSkipsSoftDeleted() {
insertMetalake(1L, "metalake1");
insertGroup(31L, "group31", 1L);
- groupMetaMapper.softDeleteGroupMetaByGroupId(31L);
+ groupMetaMapper.softDeleteGroupMetaByGroupId(31L, 1L);
long beforeUpdatedAt = queryUpdatedAt("group_meta", "group_id", 31L);
groupMetaMapper.touchGroupUpdatedAt(31L);
@@ -287,6 +287,16 @@ public class TestAuthMappers {
Assertions.assertEquals(expected, info.getUpdatedAt());
}
+ @Test
+ void testGroupDeleteUsesCurrentVersion() {
+ insertMetalake(1L, "metalake1");
+ insertGroup(33L, "group33", 1L);
+
+ Assertions.assertEquals(0,
groupMetaMapper.softDeleteGroupMetaByGroupId(33L, 2L));
+
Assertions.assertNotNull(groupMetaMapper.selectGroupMetaByMetalakeIdAndName(1L,
"group33"));
+ Assertions.assertEquals(1,
groupMetaMapper.softDeleteGroupMetaByGroupId(33L, 1L));
+ }
+
@Test
void testUserDeleteUsesCurrentVersion() {
insertMetalake(1L, "metalake1");
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestGroupMetaService.java
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestGroupMetaService.java
index e14f91a4c3..90f850c709 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestGroupMetaService.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestGroupMetaService.java
@@ -44,6 +44,7 @@ import org.apache.gravitino.Namespace;
import org.apache.gravitino.authorization.AuthorizationUtils;
import org.apache.gravitino.authorization.PagedResult;
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.GroupEntity;
@@ -51,7 +52,10 @@ import org.apache.gravitino.meta.RoleEntity;
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.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.auth.GroupUpdatedAt;
import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
@@ -709,8 +713,8 @@ class TestGroupMetaService extends TestJDBCBackend {
Assertions.assertEquals("creator", grantRevokeGroup.auditInfo().creator());
Assertions.assertEquals("grantRevokeUser",
grantRevokeGroup.auditInfo().lastModifier());
- // no update
- Function<GroupEntity, GroupEntity> noUpdater =
+ // metadata-only update
+ Function<GroupEntity, GroupEntity> metadataUpdater =
group -> {
AuditInfo updateAuditInfo =
AuditInfo.builder()
@@ -732,19 +736,20 @@ class TestGroupMetaService extends TestJDBCBackend {
.withAuditInfo(updateAuditInfo)
.build();
};
- long beforeNoUpdate = getGroupUpdatedAt(group1.name()).getUpdatedAt();
-
Assertions.assertNotNull(groupMetaService.updateGroup(group1.nameIdentifier(),
noUpdater));
- Assertions.assertEquals(beforeNoUpdate,
getGroupUpdatedAt(group1.name()).getUpdatedAt());
- GroupEntity noUpdaterGroup =
+ long beforeMetadataUpdate =
getGroupUpdatedAt(group1.name()).getUpdatedAt();
+ Assertions.assertNotNull(
+ groupMetaService.updateGroup(group1.nameIdentifier(),
metadataUpdater));
+ Assertions.assertTrue(getGroupUpdatedAt(group1.name()).getUpdatedAt() >=
beforeMetadataUpdate);
+ GroupEntity metadataUpdatedGroup =
GroupMetaService.getInstance().getGroupByIdentifier(group1.nameIdentifier());
- Assertions.assertEquals(group1.id(), noUpdaterGroup.id());
- Assertions.assertEquals(group1.name(), noUpdaterGroup.name());
+ Assertions.assertEquals(group1.id(), metadataUpdatedGroup.id());
+ Assertions.assertEquals(group1.name(), metadataUpdatedGroup.name());
Assertions.assertEquals(
- Sets.newHashSet("role1", "role4"),
Sets.newHashSet(noUpdaterGroup.roleNames()));
+ Sets.newHashSet("role1", "role4"),
Sets.newHashSet(metadataUpdatedGroup.roleNames()));
Assertions.assertEquals(
- Sets.newHashSet(role1.id(), role4.id()),
Sets.newHashSet(noUpdaterGroup.roleIds()));
- Assertions.assertEquals("creator", noUpdaterGroup.auditInfo().creator());
- Assertions.assertEquals("grantRevokeUser",
noUpdaterGroup.auditInfo().lastModifier());
+ Sets.newHashSet(role1.id(), role4.id()),
Sets.newHashSet(metadataUpdatedGroup.roleIds()));
+ Assertions.assertEquals("creator",
metadataUpdatedGroup.auditInfo().creator());
+ Assertions.assertEquals("noUpdateUser",
metadataUpdatedGroup.auditInfo().lastModifier());
// Delete a role, the group entity won't contain this role.
RoleMetaService.getInstance().deleteRole(role1.nameIdentifier());
@@ -1081,6 +1086,64 @@ class TestGroupMetaService extends TestJDBCBackend {
IllegalArgumentException.class, () ->
svc.getGroupByExternalId(groupExtIdent("")));
}
+ @TestTemplate
+ void testConcurrentUpdateDoesNotChangeRolesOnConflict() throws IOException {
+ createAndInsertMakeLake(metalakeName);
+ createAndInsertCatalog(metalakeName, catalogName);
+ RoleEntity role1 =
+ createRoleEntity(
+ RandomIdGenerator.INSTANCE.nextId(),
+ AuthorizationUtils.ofRoleNamespace(metalakeName),
+ "role1",
+ AUDIT_INFO,
+ catalogName);
+ RoleEntity role2 =
+ createRoleEntity(
+ RandomIdGenerator.INSTANCE.nextId(),
+ AuthorizationUtils.ofRoleNamespace(metalakeName),
+ "role2",
+ AUDIT_INFO,
+ catalogName);
+ RoleMetaService.getInstance().insertRole(role1, false);
+ RoleMetaService.getInstance().insertRole(role2, false);
+ GroupEntity group =
+ createGroupEntity(
+ RandomIdGenerator.INSTANCE.nextId(),
+ AuthorizationUtils.ofGroupNamespace(metalakeName),
+ "concurrent-group",
+ AUDIT_INFO,
+ Lists.newArrayList(role1.name()),
+ Lists.newArrayList(role1.id()));
+ GroupMetaService.getInstance().insertGroup(group, false);
+
+ Assertions.assertThrows(
+ OptimisticLockException.class,
+ () ->
+ GroupMetaService.getInstance()
+ .updateGroup(
+ group.nameIdentifier(),
+ (GroupEntity oldGroup) -> {
+ advanceGroupVersion(group.id());
+ List<String> roleNames =
Lists.newArrayList(oldGroup.roleNames());
+ List<Long> roleIds =
Lists.newArrayList(oldGroup.roleIds());
+ roleNames.add(role2.name());
+ roleIds.add(role2.id());
+ return GroupEntity.builder()
+ .withId(oldGroup.id())
+ .withName(oldGroup.name())
+ .withNamespace(oldGroup.namespace())
+ .withExternalId(oldGroup.externalId())
+ .withRoleNames(roleNames)
+ .withRoleIds(roleIds)
+ .withAuditInfo(oldGroup.auditInfo())
+ .build();
+ }));
+
+ GroupEntity storedGroup =
+
GroupMetaService.getInstance().getGroupByIdentifier(group.nameIdentifier());
+ assertEquals(Sets.newHashSet(role1.id()),
Sets.newHashSet(storedGroup.roleIds()));
+ }
+
@TestTemplate
void testExtDup() throws IOException {
GroupMetaService svc = groupMetaService();
@@ -1090,6 +1153,130 @@ class TestGroupMetaService extends TestJDBCBackend {
() -> svc.insertGroup(groupWithExtId("g2", "ext-1"), false));
}
+ @TestTemplate
+ void testCreateLocksMetalakeWithoutChangingVersion() throws IOException {
+ createAndInsertMakeLake(metalakeName);
+ GroupMetaService service = GroupMetaService.getInstance();
+ MetalakePO beforeCreate = getMetalakePO();
+ GroupEntity group = groupWithExtId("fenced-group", "fenced-group-ext-id");
+
+ service.insertGroup(group, false);
+
+ MetalakePO afterCreate = getMetalakePO();
+ assertEquals(beforeCreate.getCurrentVersion(),
afterCreate.getCurrentVersion());
+ assertEquals(beforeCreate.getLastVersion(), afterCreate.getLastVersion());
+
+ GroupEntity duplicate = groupWithExtId(group.name(), "another-ext-id");
+ Assertions.assertThrows(
+ EntityAlreadyExistsException.class, () ->
service.insertGroup(duplicate, false));
+
+ MetalakePO afterFailedCreate = getMetalakePO();
+ assertEquals(afterCreate.getCurrentVersion(),
afterFailedCreate.getCurrentVersion());
+ assertEquals(afterCreate.getLastVersion(),
afterFailedCreate.getLastVersion());
+ }
+
+ @TestTemplate
+ void testOverwriteInsertAdvancesVersion() throws IOException {
+ GroupMetaService service = groupMetaService();
+ GroupEntity group = groupWithExtId("overwrite-group",
"overwrite-group-ext-id");
+ service.insertGroup(group, false);
+ GroupPO initialPO = getGroupPO(group.name());
+
+ service.insertGroup(group, true);
+
+ GroupPO overwrittenPO = getGroupPO(group.name());
+ assertEquals(initialPO.getCurrentVersion() + 1,
overwrittenPO.getCurrentVersion());
+ assertEquals(overwrittenPO.getCurrentVersion(),
overwrittenPO.getLastVersion());
+ int staleDelete =
+ SessionUtils.doWithCommitAndFetchResult(
+ GroupMetaMapper.class,
+ mapper ->
+ mapper.softDeleteGroupMetaByGroupId(group.id(),
initialPO.getCurrentVersion()));
+ assertEquals(0, staleDelete);
+ }
+
+ @TestTemplate
+ void testMetadataOnlyUpdateUsesOcc() throws IOException {
+ GroupMetaService service = groupMetaService();
+ GroupEntity group = groupWithExtId("metadata-only-group",
"metadata-only-ext-id");
+ service.insertGroup(group, false);
+ GroupPO beforeUpdate = getGroupPO(group.name());
+
+ service.updateGroup(
+ group.nameIdentifier(), (GroupEntity oldGroup) -> copyGroup(oldGroup,
"updated-ext-id"));
+
+ GroupPO afterUpdate = getGroupPO(group.name());
+ assertEquals(beforeUpdate.getCurrentVersion() + 1,
afterUpdate.getCurrentVersion());
+ assertEquals(
+ "updated-ext-id",
service.getGroupByIdentifier(group.nameIdentifier()).externalId());
+
+ Assertions.assertThrows(
+ OptimisticLockException.class,
+ () ->
+ service.updateGroup(
+ group.nameIdentifier(),
+ (GroupEntity oldGroup) -> {
+ advanceGroupVersion(group.id());
+ return copyGroup(oldGroup, "conflicting-ext-id");
+ }));
+ assertEquals(
+ "updated-ext-id",
service.getGroupByIdentifier(group.nameIdentifier()).externalId());
+ }
+
+ @TestTemplate
+ void testStaleDeleteReportsConflict() throws IOException {
+ GroupMetaService service = groupMetaService();
+ GroupEntity group = groupWithExtId("stale-delete-group",
"stale-delete-group-ext-id");
+ service.insertGroup(group, false);
+ GroupPO staleGroupPO = getGroupPO(group.name());
+ advanceGroupVersion(group.id());
+
+ Assertions.assertThrows(
+ OptimisticLockException.class,
+ () -> service.deleteGroupWithVersion(group.nameIdentifier(),
staleGroupPO));
+ assertEquals(group.id(),
service.getGroupByIdentifier(group.nameIdentifier()).id());
+ }
+
+ @TestTemplate
+ void testAlterReportsNoSuchWhenGroupIsDeletedConcurrently() throws
IOException {
+ GroupMetaService service = groupMetaService();
+ GroupEntity group = groupWithExtId("deleted-during-alter",
"deleted-during-alter-ext-id");
+ service.insertGroup(group, false);
+
+ Assertions.assertThrows(
+ NoSuchEntityException.class,
+ () ->
+ service.updateGroup(
+ group.nameIdentifier(),
+ (GroupEntity oldGroup) -> {
+ service.deleteGroup(group.nameIdentifier());
+ return copyGroup(oldGroup, "ignored-ext-id");
+ }));
+ }
+
+ @TestTemplate
+ void testByIdWritesUseOcc() throws IOException {
+ GroupMetaService service = groupMetaService();
+ GroupEntity group = groupWithExtId("by-id-group", "by-id-group-ext-id");
+ service.insertGroup(group, false);
+
+ Assertions.assertThrows(
+ OptimisticLockException.class,
+ () ->
+ service.updateGroupById(
+ metalakeName,
+ group.id(),
+ (GroupEntity oldGroup) -> {
+ advanceGroupVersion(group.id());
+ return copyGroup(oldGroup, "conflicting-by-id-ext-id");
+ }));
+ assertEquals(
+ "by-id-group-ext-id",
service.getGroupByIdentifier(group.nameIdentifier()).externalId());
+
+ assertTrue(service.deleteGroupById(metalakeName, group.id()));
+ assertFalse(service.deleteGroupById(metalakeName, group.id()));
+ }
+
@TestTemplate
void testGroupExtDel() throws IOException {
GroupMetaService svc = groupMetaService();
@@ -1240,6 +1427,30 @@ class TestGroupMetaService extends TestJDBCBackend {
return GroupMetaService.getInstance();
}
+ private MetalakePO getMetalakePO() {
+ return SessionUtils.getWithoutCommit(
+ MetalakeMetaMapper.class, mapper ->
mapper.selectMetalakeMetaByName(metalakeName));
+ }
+
+ private GroupPO getGroupPO(String groupName) {
+ MetalakePO metalakePO = getMetalakePO();
+ return SessionUtils.getWithoutCommit(
+ GroupMetaMapper.class,
+ mapper ->
mapper.selectGroupMetaByMetalakeIdAndName(metalakePO.getMetalakeId(),
groupName));
+ }
+
+ private GroupEntity copyGroup(GroupEntity group, String externalId) {
+ return GroupEntity.builder()
+ .withId(group.id())
+ .withName(group.name())
+ .withNamespace(group.namespace())
+ .withExternalId(externalId)
+ .withRoleNames(group.roleNames())
+ .withRoleIds(group.roleIds())
+ .withAuditInfo(group.auditInfo())
+ .build();
+ }
+
private void assertThrowsExt(Class<? extends Exception> type, Executable
executable) {
Assertions.assertThrows(type, executable);
}
@@ -1265,4 +1476,19 @@ class TestGroupMetaService extends TestJDBCBackend {
.withAuditInfo(auditInfo)
.build();
}
+
+ private void advanceGroupVersion(long groupId) {
+ try (SqlSession sqlSession =
+
SqlSessionFactoryHelper.getInstance().getSqlSessionFactory().openSession(true);
+ Connection connection = sqlSession.getConnection();
+ Statement statement = connection.createStatement()) {
+ assertEquals(
+ 1,
+ statement.executeUpdate(
+ "UPDATE group_meta SET current_version = current_version + 1
WHERE group_id = "
+ + groupId));
+ } catch (SQLException e) {
+ throw new RuntimeException("Advance group 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 8e1c563073..afb28a80f9 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
@@ -51,6 +51,7 @@ import org.apache.gravitino.meta.BaseMetalake;
import org.apache.gravitino.meta.CatalogEntity;
import org.apache.gravitino.meta.ColumnEntity;
import org.apache.gravitino.meta.FilesetEntity;
+import org.apache.gravitino.meta.GroupEntity;
import org.apache.gravitino.meta.ModelEntity;
import org.apache.gravitino.meta.ModelVersionEntity;
import org.apache.gravitino.meta.PolicyEntity;
@@ -80,6 +81,7 @@ import org.apache.gravitino.storage.relational.po.CatalogPO;
import org.apache.gravitino.storage.relational.po.ColumnPO;
import org.apache.gravitino.storage.relational.po.FilesetPO;
import org.apache.gravitino.storage.relational.po.FilesetVersionPO;
+import org.apache.gravitino.storage.relational.po.GroupPO;
import org.apache.gravitino.storage.relational.po.MetalakePO;
import org.apache.gravitino.storage.relational.po.ModelPO;
import org.apache.gravitino.storage.relational.po.ModelVersionAliasRelPO;
@@ -843,6 +845,32 @@ public class TestPOConverters {
assertEquals(8, updatePO3.getFilesetVersionPOs().get(0).getVersion());
}
+ @Test
+ public void testUpdateGroupPOVersionUsesCurrentVersion() {
+ AuditInfo auditInfo =
+
AuditInfo.builder().withCreator("creator").withCreateTime(FIX_INSTANT).build();
+ GroupEntity group =
+
GroupEntity.builder().withId(2L).withName("group").withAuditInfo(auditInfo).build();
+ GroupPO initialGroupPO =
+ POConverters.initializeGroupPOWithVersion(group,
GroupPO.builder().withMetalakeId(1L));
+ GroupPO groupPO =
+ GroupPO.builder()
+ .withGroupId(initialGroupPO.getGroupId())
+ .withGroupName(initialGroupPO.getGroupName())
+ .withMetalakeId(initialGroupPO.getMetalakeId())
+ .withExternalId(initialGroupPO.getExternalId())
+ .withAuditInfo(initialGroupPO.getAuditInfo())
+ .withCurrentVersion(7L)
+ .withLastVersion(3L)
+ .withDeletedAt(initialGroupPO.getDeletedAt())
+ .build();
+
+ GroupPO updatedGroupPO = POConverters.updateGroupPOWithVersion(groupPO,
group);
+
+ assertEquals(8, updatedGroupPO.getCurrentVersion());
+ assertEquals(8, updatedGroupPO.getLastVersion());
+ }
+
@Test
public void testUpdateUserPOVersionUsesCurrentVersion() {
AuditInfo auditInfo =
