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 ff3a9e7655 [#12739] improvement(core): add OCC for user writes (#12744)
ff3a9e7655 is described below
commit ff3a9e765522bd8c8b8e36efdd5c5dfae846c457
Author: Qi Yu <[email protected]>
AuthorDate: Tue Sep 1 08:52:11 2026 +0800
[#12739] improvement(core): add OCC for user writes (#12744)
### What changes were proposed in this pull request?
Add version-CAS optimistic concurrency control and atomic service
operations for managed users.
- Increment the user version on every update and match updates by user
ID, expected version, and active-row state.
- Make soft deletes version checked and classify a zero-row result as
either a missing user or an OCC conflict.
- Execute the user CAS before user-role and ownership mutations in one
transaction.
- Fence the parent metalake while creating a user and advance the
version on overwrite upserts.
- Add coverage for metadata-only updates, external-ID and ID updates,
stale deletes, overwrite versioning, parent fencing, and relationship
rollback.
### Why are the changes needed?
Concurrent user mutations could otherwise overwrite each other or leave
partially updated relationship state.
Fix: #12739
### Does this PR introduce _any_ user-facing change?
Concurrent managed-user 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 TestUserMetaService --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/UserMetaMapper.java | 12 +-
.../mapper/UserMetaSQLProviderFactory.java | 10 +-
.../provider/base/UserMetaBaseSQLProvider.java | 32 ++-
.../postgresql/UserMetaPostgreSQLProvider.java | 14 +-
.../relational/service/UserMetaService.java | 232 ++++++++++++++----
.../storage/relational/utils/POConverters.java | 7 +-
.../mapper/provider/base/TestAuthMappers.java | 12 +-
.../relational/service/TestUserMetaService.java | 272 ++++++++++++++++++++-
.../storage/relational/utils/TestPOConverters.java | 29 +++
9 files changed, 542 insertions(+), 78 deletions(-)
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/UserMetaMapper.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/UserMetaMapper.java
index ff503326bb..dcf0800e26 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/UserMetaMapper.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/UserMetaMapper.java
@@ -54,6 +54,10 @@ public interface UserMetaMapper {
UserPO selectUserMetaByMetalakeIdAndName(
@Param("metalakeId") Long metalakeId, @Param("userName") String name);
+ /** Returns and locks an active user by ID for the current transaction. */
+ @SelectProvider(type = UserMetaSQLProviderFactory.class, method =
"selectUserMetaByIdForUpdate")
+ UserPO selectUserMetaByIdForUpdate(@Param("userId") Long userId);
+
@InsertProvider(type = UserMetaSQLProviderFactory.class, method =
"insertUserMeta")
void insertUserMeta(@Param("userMeta") UserPO userPO);
@@ -81,8 +85,14 @@ public interface UserMetaMapper {
method = "insertUserMetaOnDuplicateKeyUpdate")
void insertUserMetaOnDuplicateKeyUpdate(@Param("userMeta") UserPO userPO);
+ /**
+ * Soft-deletes an active user only when its OCC version still matches.
+ *
+ * @return the number of deleted rows
+ */
@UpdateProvider(type = UserMetaSQLProviderFactory.class, method =
"softDeleteUserMetaByUserId")
- void softDeleteUserMetaByUserId(@Param("userId") Long userId);
+ Integer softDeleteUserMetaByUserId(
+ @Param("userId") Long userId, @Param("currentVersion") Long
currentVersion);
@UpdateProvider(
type = UserMetaSQLProviderFactory.class,
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/UserMetaSQLProviderFactory.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/UserMetaSQLProviderFactory.java
index 57b4afa3cb..ff40d94dfa 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/UserMetaSQLProviderFactory.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/UserMetaSQLProviderFactory.java
@@ -61,6 +61,11 @@ public class UserMetaSQLProviderFactory {
return getProvider().selectUserMetaByMetalakeIdAndName(metalakeId, name);
}
+ /** Returns SQL that selects and locks an active user by ID. */
+ public static String selectUserMetaByIdForUpdate(@Param("userId") Long
userId) {
+ return getProvider().selectUserMetaByIdForUpdate(userId);
+ }
+
public static String insertUserMeta(@Param("userMeta") UserPO userPO) {
return getProvider().insertUserMeta(userPO);
}
@@ -69,8 +74,9 @@ public class UserMetaSQLProviderFactory {
return getProvider().insertUserMetaOnDuplicateKeyUpdate(userPO);
}
- public static String softDeleteUserMetaByUserId(@Param("userId") Long
userId) {
- return getProvider().softDeleteUserMetaByUserId(userId);
+ public static String softDeleteUserMetaByUserId(
+ @Param("userId") Long userId, @Param("currentVersion") Long
currentVersion) {
+ return getProvider().softDeleteUserMetaByUserId(userId, currentVersion);
}
public static String softDeleteUserMetasByMetalakeId(@Param("metalakeId")
Long metalakeId) {
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/UserMetaBaseSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/UserMetaBaseSQLProvider.java
index 387c5676e0..0b5b484f22 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/UserMetaBaseSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/UserMetaBaseSQLProvider.java
@@ -53,6 +53,17 @@ public class UserMetaBaseSQLProvider {
+ " AND deleted_at = 0";
}
+ /** Returns SQL that selects and locks an active user by ID. */
+ public String selectUserMetaByIdForUpdate(@Param("userId") Long userId) {
+ return "SELECT user_id as userId, user_name as userName,"
+ + " metalake_id as metalakeId, external_id as externalId, enabled as
enabled,"
+ + " audit_info as auditInfo, current_version as currentVersion,"
+ + " last_version as lastVersion, deleted_at as deletedAt"
+ + " FROM "
+ + USER_TABLE_NAME
+ + " WHERE user_id = #{userId} AND deleted_at = 0 FOR UPDATE";
+ }
+
public String selectUserMetaByMetalakeNameAndExternalId(
@Param("metalakeName") String metalakeName, @Param("externalId") String
externalId) {
return "SELECT ut.user_id as userId, ut.user_name as userName,"
@@ -99,11 +110,8 @@ public class UserMetaBaseSQLProvider {
+ " current_version = #{newUserMeta.currentVersion},"
+ " last_version = #{newUserMeta.lastVersion},"
+ " deleted_at = #{newUserMeta.deletedAt}"
- + " WHERE external_id = #{oldUserMeta.externalId}"
- + " AND metalake_id = #{oldUserMeta.metalakeId}"
- + " AND audit_info = #{oldUserMeta.auditInfo}"
+ + " WHERE user_id = #{oldUserMeta.userId}"
+ " AND current_version = #{oldUserMeta.currentVersion}"
- + " AND last_version = #{oldUserMeta.lastVersion}"
+ " AND deleted_at = 0";
}
@@ -147,17 +155,21 @@ public class UserMetaBaseSQLProvider {
+ " audit_info = #{userMeta.auditInfo},"
+ " external_id = #{userMeta.externalId},"
+ " enabled = #{userMeta.enabled},"
- + " current_version = #{userMeta.currentVersion},"
- + " last_version = #{userMeta.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 = #{userMeta.deletedAt}";
}
- public String softDeleteUserMetaByUserId(@Param("userId") Long userId) {
+ public String softDeleteUserMetaByUserId(
+ @Param("userId") Long userId, @Param("currentVersion") Long
currentVersion) {
return "UPDATE "
+ USER_TABLE_NAME
+ " SET deleted_at = (UNIX_TIMESTAMP() * 1000.0)"
+ " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000"
- + " WHERE user_id = #{userId} AND deleted_at = 0";
+ + " WHERE user_id = #{userId}"
+ + " AND current_version = #{currentVersion} AND deleted_at = 0";
}
public String softDeleteUserMetasByMetalakeId(@Param("metalakeId") Long
metalakeId) {
@@ -181,11 +193,7 @@ public class UserMetaBaseSQLProvider {
+ " last_version = #{newUserMeta.lastVersion},"
+ " deleted_at = #{newUserMeta.deletedAt}"
+ " WHERE user_id = #{oldUserMeta.userId}"
- + " AND user_name = #{oldUserMeta.userName}"
- + " AND metalake_id = #{oldUserMeta.metalakeId}"
- + " AND audit_info = #{oldUserMeta.auditInfo}"
+ " AND current_version = #{oldUserMeta.currentVersion}"
- + " AND last_version = #{oldUserMeta.lastVersion}"
+ " AND deleted_at = 0";
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/UserMetaPostgreSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/UserMetaPostgreSQLProvider.java
index a454efa6c6..bfd53c73ef 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/UserMetaPostgreSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/UserMetaPostgreSQLProvider.java
@@ -29,11 +29,12 @@ import org.apache.ibatis.annotations.Param;
public class UserMetaPostgreSQLProvider extends UserMetaBaseSQLProvider {
@Override
- public String softDeleteUserMetaByUserId(Long userId) {
+ public String softDeleteUserMetaByUserId(Long userId, Long currentVersion) {
return "UPDATE "
+ USER_TABLE_NAME
+ " SET deleted_at = CAST(EXTRACT(EPOCH FROM CURRENT_TIMESTAMP) * 1000
AS BIGINT)"
- + " WHERE user_id = #{userId} AND deleted_at = 0";
+ + " WHERE user_id = #{userId}"
+ + " AND current_version = #{currentVersion} AND deleted_at = 0";
}
@Override
@@ -68,8 +69,13 @@ public class UserMetaPostgreSQLProvider extends
UserMetaBaseSQLProvider {
+ " external_id = #{userMeta.externalId},"
+ " enabled = #{userMeta.enabled},"
+ " audit_info = #{userMeta.auditInfo},"
- + " current_version = #{userMeta.currentVersion},"
- + " last_version = #{userMeta.lastVersion},"
+ // PostgreSQL requires the stored-row column to be qualified in ON
CONFLICT assignments.
+ + " current_version = "
+ + USER_TABLE_NAME
+ + ".current_version + 1,"
+ + " last_version = "
+ + USER_TABLE_NAME
+ + ".current_version + 1,"
+ " deleted_at = #{userMeta.deletedAt}";
}
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 3eacd9850f..2ba8d38c9e 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
@@ -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;
@@ -41,10 +42,12 @@ import
org.apache.gravitino.exceptions.NoSuchEntityException;
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.MetalakeMetaMapper;
import org.apache.gravitino.storage.relational.mapper.OwnerMetaMapper;
import org.apache.gravitino.storage.relational.mapper.UserMetaMapper;
import org.apache.gravitino.storage.relational.mapper.UserRoleRelMapper;
import org.apache.gravitino.storage.relational.po.ExtendedUserPO;
+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.po.UserRoleRelPO;
@@ -132,9 +135,18 @@ public class UserMetaService {
try {
AuthorizationUtils.checkUser(userEntity.nameIdentifier());
- Long metalakeId =
-
MetalakeMetaService.getInstance().getMetalakeIdByName(userEntity.namespace().level(0));
- UserPO.Builder builder = UserPO.builder().withMetalakeId(metalakeId);
+ String metalakeName = userEntity.namespace().level(0);
+ MetalakePO metalakePO =
+ SessionUtils.getWithoutCommit(
+ MetalakeMetaMapper.class, mapper ->
mapper.selectMetalakeMetaByName(metalakeName));
+ if (metalakePO == null) {
+ throw new NoSuchEntityException(
+ NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
+ Entity.EntityType.METALAKE.name().toLowerCase(),
+ metalakeName);
+ }
+
+ UserPO.Builder builder =
UserPO.builder().withMetalakeId(metalakePO.getMetalakeId());
UserPO userPO = POConverters.initializeUserPOWithVersion(userEntity,
builder);
List<Long> roleIds =
Optional.ofNullable(userEntity.roleIds()).orElse(Lists.newArrayList());
@@ -142,6 +154,7 @@ public class UserMetaService {
POConverters.initializeUserRoleRelsPOWithVersion(userEntity,
roleIds);
SessionUtils.doMultipleWithCommit(
+ () -> lockMetalakeForUserCreate(metalakePO),
() ->
SessionUtils.doWithoutCommit(
UserMetaMapper.class,
@@ -175,12 +188,36 @@ public class UserMetaService {
public boolean deleteUser(NameIdentifier identifier) {
AuthorizationUtils.checkUser(identifier);
- Long userId = EntityIdService.getEntityId(identifier,
Entity.EntityType.USER);
+ Long metalakeId =
+
MetalakeMetaService.getInstance().getMetalakeIdByName(identifier.namespace().level(0));
+ UserPO userPO = getUserPOByMetalakeIdAndName(metalakeId,
identifier.name());
+
+ deleteUserWithVersion(identifier, userPO);
+ return true;
+ }
+ /**
+ * Deletes the user whose version matches {@code observedUserPO}, 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 #deleteUser(NameIdentifier)}, which reads the row
first.
+ *
+ * @param identifier the user being deleted, used only to build the error
+ * @param observedUserPO the user row the caller observed, carrying the
version to match
+ */
+ void deleteUserWithVersion(NameIdentifier identifier, UserPO observedUserPO)
{
+ Long userId = observedUserPO.getUserId();
SessionUtils.doMultipleWithCommit(
- () ->
- SessionUtils.doWithoutCommit(
- UserMetaMapper.class, mapper ->
mapper.softDeleteUserMetaByUserId(userId)),
+ () -> {
+ int deleted =
+ SessionUtils.getWithoutCommit(
+ UserMetaMapper.class,
+ mapper ->
+ mapper.softDeleteUserMetaByUserId(
+ userId, observedUserPO.getCurrentVersion()));
+ if (deleted == 0) {
+ throw userWriteFailure(identifier, observedUserPO,
UserLookup.NAME);
+ }
+ },
() ->
SessionUtils.doWithoutCommit(
UserRoleRelMapper.class, mapper ->
mapper.softDeleteUserRoleRelByUserId(userId)),
@@ -190,7 +227,6 @@ public class UserMetaService {
mapper ->
mapper.softDeleteOwnerRelByOwnerIdAndType(
userId, Entity.EntityType.USER.name())));
- return true;
}
@Monitored(metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
baseMetricName = "updateUser")
@@ -221,18 +257,23 @@ public class UserMetaService {
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(
- UserMetaMapper.class,
- mapper ->
- mapper.updateUserMeta(
- POConverters.updateUserPOWithVersion(oldUserPO,
newEntity), oldUserPO)),
+ () -> {
+ int updated =
+ SessionUtils.getWithoutCommit(
+ UserMetaMapper.class,
+ mapper ->
+ mapper.updateUserMeta(
+ POConverters.updateUserPOWithVersion(oldUserPO,
newEntity), oldUserPO));
+ if (updated == 0) {
+ throw userWriteFailure(identifier, oldUserPO, UserLookup.NAME);
+ }
+ },
() -> {
if (insertRoleIds.isEmpty()) {
return;
@@ -371,12 +412,17 @@ public class UserMetaService {
try {
SessionUtils.doMultipleWithCommit(
- () ->
- SessionUtils.doWithoutCommit(
- UserMetaMapper.class,
- mapper ->
- mapper.updateUserMetaByExternalId(
- POConverters.updateUserPOWithVersion(oldUserPO,
newEntity), oldUserPO)),
+ () -> {
+ int updated =
+ SessionUtils.getWithoutCommit(
+ UserMetaMapper.class,
+ mapper ->
+ mapper.updateUserMetaByExternalId(
+ POConverters.updateUserPOWithVersion(oldUserPO,
newEntity), oldUserPO));
+ if (updated == 0) {
+ throw userWriteFailure(ident, oldUserPO, UserLookup.EXTERNAL_ID);
+ }
+ },
() ->
SessionUtils.doWithoutCommit(
UserMetaMapper.class,
@@ -430,12 +476,19 @@ public class UserMetaService {
try {
SessionUtils.doMultipleWithCommit(
- () ->
- SessionUtils.doWithoutCommit(
- UserMetaMapper.class,
- mapper ->
- mapper.updateUserMeta(
- POConverters.updateUserPOWithVersion(oldUserPO,
newEntity), oldUserPO)),
+ () -> {
+ int updated =
+ SessionUtils.getWithoutCommit(
+ UserMetaMapper.class,
+ mapper ->
+ mapper.updateUserMeta(
+ POConverters.updateUserPOWithVersion(oldUserPO,
newEntity), oldUserPO));
+ if (updated == 0) {
+ NameIdentifier userIdIdentifier =
+ AuthorizationUtils.ofUser(metalake, String.valueOf(userId));
+ throw userWriteFailure(userIdIdentifier, oldUserPO,
UserLookup.ID);
+ }
+ },
() ->
SessionUtils.doWithoutCommit(
UserMetaMapper.class,
@@ -452,26 +505,42 @@ public class UserMetaService {
metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
baseMetricName = "deleteUserById")
public boolean deleteUserById(String metalake, long userId) {
+ UserPO userPO;
try {
- getUserPOByMetalakeNameAndId(metalake, userId);
+ userPO = getUserPOByMetalakeNameAndId(metalake, userId);
} catch (NoSuchEntityException e) {
return false;
}
+ NameIdentifier identifier = AuthorizationUtils.ofUser(metalake,
userPO.getUserName());
+ // 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 deletedUser = new AtomicBoolean(false);
SessionUtils.doMultipleWithCommit(
- () ->
- SessionUtils.doWithoutCommit(
- UserMetaMapper.class, mapper ->
mapper.softDeleteUserMetaByUserId(userId)),
- () ->
- SessionUtils.doWithoutCommit(
- UserRoleRelMapper.class, mapper ->
mapper.softDeleteUserRoleRelByUserId(userId)),
- () ->
- SessionUtils.doWithoutCommit(
- OwnerMetaMapper.class,
- mapper ->
- mapper.softDeleteOwnerRelByOwnerIdAndType(
- userId, Entity.EntityType.USER.name())));
- return true;
+ () -> {
+ int deleted =
+ SessionUtils.getWithoutCommit(
+ UserMetaMapper.class,
+ mapper -> mapper.softDeleteUserMetaByUserId(userId,
userPO.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 (getUserPOByIdForUpdate(userId) == null) {
+ return;
+ }
+ throw
ExceptionUtils.concurrentModification(Entity.EntityType.USER, identifier);
+ }
+
+ deletedUser.set(true);
+ SessionUtils.doWithoutCommit(
+ UserRoleRelMapper.class, mapper ->
mapper.softDeleteUserRoleRelByUserId(userId));
+ SessionUtils.doWithoutCommit(
+ OwnerMetaMapper.class,
+ mapper ->
+ mapper.softDeleteOwnerRelByOwnerIdAndType(userId,
Entity.EntityType.USER.name()));
+ });
+ return deletedUser.get();
}
@Monitored(
@@ -511,4 +580,79 @@ public class UserMetaService {
.collect(Collectors.toList());
return new PagedResult<>(totalCount, users);
}
+
+ /**
+ * Holds the parent metalake row for the rest of the transaction, so the
user cannot be created
+ * under a metalake that is going away.
+ *
+ * <p>The lock is shared, not exclusive: many users 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 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());
+ }
+ }
+
+ private RuntimeException userWriteFailure(
+ NameIdentifier identifier, UserPO observedUserPO, UserLookup 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.
+ 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);
+ }
+
+ private UserPO getUserPOByIdForUpdate(long userId) {
+ return SessionUtils.getWithoutCommit(
+ UserMetaMapper.class, mapper ->
mapper.selectUserMetaByIdForUpdate(userId));
+ }
+
+ /**
+ * How the caller addressed the user, which decides what counts as "the same
user" when a failed
+ * compare-and-set is classified. A caller that used the name is looking for
that name, so a
+ * rename means the user it asked for is gone; the same holds for the
external ID. A caller that
+ * used the ID addressed the row itself, so a rename leaves it addressing
the same user and only
+ * the metalake has to still match.
+ */
+ private enum UserLookup {
+ NAME,
+ EXTERNAL_ID,
+ 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 889aa0d6c4..c05cd50d9b 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
@@ -960,14 +960,11 @@ public class POConverters {
* Update UserPO version
*
* @param oldUserPO the old UserPO object
- * @param newUser the new TableEntity object
+ * @param newUser the new UserEntity object
* @return UserPO object with updated version
*/
public static UserPO updateUserPOWithVersion(UserPO oldUserPO, UserEntity
newUser) {
- Long lastVersion = oldUserPO.getLastVersion();
- // TODO: set the version to the last version + 1 when having some fields
need be multiple
- // version
- Long nextVersion = lastVersion;
+ Long nextVersion = oldUserPO.getCurrentVersion() + 1;
try {
return UserPO.builder()
.withUserId(oldUserPO.getUserId())
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 e26613c790..550ad3c54c 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
@@ -219,7 +219,7 @@ public class TestAuthMappers {
void testUserMetaTouchUpdatedAtSkipsSoftDeleted() {
insertMetalake(1L, "metalake1");
insertUser(22L, "user22", 1L);
- userMetaMapper.softDeleteUserMetaByUserId(22L);
+ userMetaMapper.softDeleteUserMetaByUserId(22L, 1L);
long beforeUpdatedAt = queryUpdatedAt("user_meta", "user_id", 22L);
userMetaMapper.touchUserUpdatedAt(22L);
@@ -287,6 +287,16 @@ public class TestAuthMappers {
Assertions.assertEquals(expected, info.getUpdatedAt());
}
+ @Test
+ void testUserDeleteUsesCurrentVersion() {
+ insertMetalake(1L, "metalake1");
+ insertUser(23L, "user23", 1L);
+
+ Assertions.assertEquals(0, userMetaMapper.softDeleteUserMetaByUserId(23L,
2L));
+
Assertions.assertNotNull(userMetaMapper.selectUserMetaByMetalakeIdAndName(1L,
"user23"));
+ Assertions.assertEquals(1, userMetaMapper.softDeleteUserMetaByUserId(23L,
1L));
+ }
+
@Test
void testOwnerMetaSelectOwnerByMetadataObjectIdAndType() {
insertMetalake(1L, "metalake1");
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestUserMetaService.java
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestUserMetaService.java
index bb377e7c38..59e72f1412 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestUserMetaService.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestUserMetaService.java
@@ -45,6 +45,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.CatalogEntity;
@@ -57,9 +58,12 @@ import org.apache.gravitino.meta.TopicEntity;
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.MetalakeMetaMapper;
import org.apache.gravitino.storage.relational.mapper.RoleMetaMapper;
import org.apache.gravitino.storage.relational.mapper.UserMetaMapper;
+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.po.auth.AuthPrefetchRow;
import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
import org.apache.gravitino.storage.relational.utils.SessionUtils;
@@ -833,7 +837,7 @@ class TestUserMetaService extends TestJDBCBackend {
Assertions.assertEquals("creator", grantRevokeUser.auditInfo().creator());
Assertions.assertEquals("grantRevokeUser",
grantRevokeUser.auditInfo().lastModifier());
- Function<UserEntity, UserEntity> noUpdater =
+ Function<UserEntity, UserEntity> metadataUpdater =
user -> {
AuditInfo updateAuditInfo =
AuditInfo.builder()
@@ -855,17 +859,17 @@ class TestUserMetaService extends TestJDBCBackend {
.withAuditInfo(updateAuditInfo)
.build();
};
-
Assertions.assertNotNull(userMetaService.updateUser(user1.nameIdentifier(),
noUpdater));
- UserEntity noUpdaterUser =
+
Assertions.assertNotNull(userMetaService.updateUser(user1.nameIdentifier(),
metadataUpdater));
+ UserEntity metadataUpdatedUser =
UserMetaService.getInstance().getUserByIdentifier(user1.nameIdentifier());
- Assertions.assertEquals(user1.id(), noUpdaterUser.id());
- Assertions.assertEquals(user1.name(), noUpdaterUser.name());
+ Assertions.assertEquals(user1.id(), metadataUpdatedUser.id());
+ Assertions.assertEquals(user1.name(), metadataUpdatedUser.name());
Assertions.assertEquals(
- Sets.newHashSet("role1", "role4"),
Sets.newHashSet(noUpdaterUser.roleNames()));
+ Sets.newHashSet("role1", "role4"),
Sets.newHashSet(metadataUpdatedUser.roleNames()));
Assertions.assertEquals(
- Sets.newHashSet(role1.id(), role4.id()),
Sets.newHashSet(noUpdaterUser.roleIds()));
- Assertions.assertEquals("creator", noUpdaterUser.auditInfo().creator());
- Assertions.assertEquals("grantRevokeUser",
noUpdaterUser.auditInfo().lastModifier());
+ Sets.newHashSet(role1.id(), role4.id()),
Sets.newHashSet(metadataUpdatedUser.roleIds()));
+ Assertions.assertEquals("creator",
metadataUpdatedUser.auditInfo().creator());
+ Assertions.assertEquals("noUpdateUser",
metadataUpdatedUser.auditInfo().lastModifier());
// Delete a role, the user entity won't contain this role.
RoleMetaService.getInstance().deleteRole(role1.nameIdentifier());
@@ -1467,11 +1471,246 @@ class TestUserMetaService extends TestJDBCBackend {
() -> svc.insertUser(userWithExtId("u2", "ext-1"), false));
}
+ @TestTemplate
+ void testCreateLocksMetalakeWithoutChangingVersion() throws IOException {
+ createAndInsertMakeLake(metalakeName);
+ UserMetaService service = UserMetaService.getInstance();
+ MetalakePO beforeCreate = getMetalakePO();
+ UserEntity user = userWithExtId("fenced-user", "fenced-user-ext-id");
+
+ service.insertUser(user, false);
+
+ MetalakePO afterCreate = getMetalakePO();
+ assertEquals(beforeCreate.getCurrentVersion(),
afterCreate.getCurrentVersion());
+ assertEquals(beforeCreate.getLastVersion(), afterCreate.getLastVersion());
+
+ UserEntity duplicate = userWithExtId(user.name(), "another-ext-id");
+ Assertions.assertThrows(
+ EntityAlreadyExistsException.class, () ->
service.insertUser(duplicate, false));
+
+ MetalakePO afterFailedCreate = getMetalakePO();
+ assertEquals(afterCreate.getCurrentVersion(),
afterFailedCreate.getCurrentVersion());
+ assertEquals(afterCreate.getLastVersion(),
afterFailedCreate.getLastVersion());
+ }
+
+ @TestTemplate
+ void testOverwriteInsertAdvancesVersion() throws IOException {
+ UserMetaService service = userMetaService();
+ UserEntity user = userWithExtId("overwrite-user", "overwrite-user-ext-id");
+ service.insertUser(user, false);
+ UserPO initialPO = getUserPO(user.name());
+
+ service.insertUser(user, true);
+
+ UserPO overwrittenPO = getUserPO(user.name());
+ assertEquals(initialPO.getCurrentVersion() + 1,
overwrittenPO.getCurrentVersion());
+ assertEquals(overwrittenPO.getCurrentVersion(),
overwrittenPO.getLastVersion());
+ int staleDelete =
+ SessionUtils.doWithCommitAndFetchResult(
+ UserMetaMapper.class,
+ mapper -> mapper.softDeleteUserMetaByUserId(user.id(),
initialPO.getCurrentVersion()));
+ assertEquals(0, staleDelete);
+ }
+
+ @TestTemplate
+ void testMetadataOnlyUpdateUsesOcc() throws IOException {
+ UserMetaService service = userMetaService();
+ UserEntity user = userWithExtId("metadata-only-user",
"metadata-only-ext-id");
+ service.insertUser(user, false);
+ UserPO beforeUpdate = getUserPO(user.name());
+
+ service.updateUser(user.nameIdentifier(), enabledUpdater(false));
+
+ UserPO afterUpdate = getUserPO(user.name());
+ assertEquals(beforeUpdate.getCurrentVersion() + 1,
afterUpdate.getCurrentVersion());
+ assertFalse(service.getUserByIdentifier(user.nameIdentifier()).enabled());
+
+ Assertions.assertThrows(
+ OptimisticLockException.class,
+ () ->
+ service.updateUser(
+ user.nameIdentifier(),
+ (UserEntity oldUser) -> {
+ advanceUserVersion(user.id());
+ return enabledUpdater(true).apply(oldUser);
+ }));
+ assertFalse(service.getUserByIdentifier(user.nameIdentifier()).enabled());
+ }
+
+ @TestTemplate
+ void testStaleDeleteReportsConflict() throws IOException {
+ UserMetaService service = userMetaService();
+ UserEntity user = userWithExtId("stale-delete-user",
"stale-delete-ext-id");
+ service.insertUser(user, false);
+ UserPO staleUserPO = getUserPO(user.name());
+ advanceUserVersion(user.id());
+
+ Assertions.assertThrows(
+ OptimisticLockException.class,
+ () -> service.deleteUserWithVersion(user.nameIdentifier(),
staleUserPO));
+ assertEquals(user.id(),
service.getUserByIdentifier(user.nameIdentifier()).id());
+ }
+
+ @TestTemplate
+ void testAlterReportsNoSuchWhenUserIsDeletedConcurrently() throws
IOException {
+ UserMetaService service = userMetaService();
+ UserEntity user = userWithExtId("deleted-during-alter",
"deleted-during-alter-ext-id");
+ service.insertUser(user, false);
+
+ Assertions.assertThrows(
+ NoSuchEntityException.class,
+ () ->
+ service.updateUser(
+ user.nameIdentifier(),
+ (UserEntity oldUser) -> {
+ service.deleteUser(user.nameIdentifier());
+ return enabledUpdater(false).apply(oldUser);
+ }));
+ }
+
+ @TestTemplate
+ void testConcurrentUpdateDoesNotChangeRolesOnConflict() throws IOException {
+ createAndInsertMakeLake(metalakeName);
+ createAndInsertCatalog(metalakeName, "catalog");
+ RoleEntity role1 =
+ createRoleEntity(
+ RandomIdGenerator.INSTANCE.nextId(),
+ AuthorizationUtils.ofRoleNamespace(metalakeName),
+ "role1",
+ AUDIT_INFO,
+ "catalog");
+ RoleEntity role2 =
+ createRoleEntity(
+ RandomIdGenerator.INSTANCE.nextId(),
+ AuthorizationUtils.ofRoleNamespace(metalakeName),
+ "role2",
+ AUDIT_INFO,
+ "catalog");
+ RoleMetaService.getInstance().insertRole(role1, false);
+ RoleMetaService.getInstance().insertRole(role2, false);
+ UserEntity user =
+ createUserEntity(
+ RandomIdGenerator.INSTANCE.nextId(),
+ AuthorizationUtils.ofUserNamespace(metalakeName),
+ "concurrent-user",
+ AUDIT_INFO,
+ Lists.newArrayList(role1.name()),
+ Lists.newArrayList(role1.id()));
+ UserMetaService.getInstance().insertUser(user, false);
+
+ Assertions.assertThrows(
+ OptimisticLockException.class,
+ () ->
+ UserMetaService.getInstance()
+ .updateUser(
+ user.nameIdentifier(),
+ (UserEntity oldUser) -> {
+ advanceUserVersion(user.id());
+ List<String> roleNames =
Lists.newArrayList(oldUser.roleNames());
+ List<Long> roleIds =
Lists.newArrayList(oldUser.roleIds());
+ roleNames.add(role2.name());
+ roleIds.add(role2.id());
+ return UserEntity.builder()
+ .withId(oldUser.id())
+ .withName(oldUser.name())
+ .withNamespace(oldUser.namespace())
+ .withExternalId(oldUser.externalId())
+ .withEnabled(oldUser.enabled())
+ .withRoleNames(roleNames)
+ .withRoleIds(roleIds)
+ .withAuditInfo(oldUser.auditInfo())
+ .build();
+ }));
+
+ UserEntity storedUser =
+
UserMetaService.getInstance().getUserByIdentifier(user.nameIdentifier());
+ assertEquals(Sets.newHashSet(role1.id()),
Sets.newHashSet(storedUser.roleIds()));
+ }
+
+ @TestTemplate
+ void testConcurrentExternalIdUpdateRollsBackOnConflict() throws IOException {
+ UserMetaService service = userMetaService();
+ UserEntity user = userWithExtId("concurrent-user", "concurrent-ext-id");
+ service.insertUser(user, false);
+
+ Assertions.assertThrows(
+ OptimisticLockException.class,
+ () ->
+ service.updateUserByExternalId(
+ userExtIdent(user.externalId()),
+ (UserEntity oldUser) -> {
+ advanceUserVersion(user.id());
+ return UserEntity.builder()
+ .withId(oldUser.id())
+ .withName(oldUser.name())
+ .withNamespace(oldUser.namespace())
+ .withExternalId(oldUser.externalId())
+ .withEnabled(false)
+ .withRoleNames(oldUser.roleNames())
+ .withRoleIds(oldUser.roleIds())
+ .withAuditInfo(oldUser.auditInfo())
+ .build();
+ }));
+
+ assertTrue(queryEnabledByExtId(user.externalId()));
+ }
+
+ @TestTemplate
+ void testConcurrentByIdUpdateRollsBackOnConflict() throws IOException {
+ UserMetaService service = userMetaService();
+ UserEntity user = userWithExtId("concurrent-by-id-user",
"concurrent-by-id-ext-id");
+ service.insertUser(user, false);
+
+ Assertions.assertThrows(
+ OptimisticLockException.class,
+ () ->
+ service.updateUserById(
+ metalakeName,
+ user.id(),
+ (UserEntity oldUser) -> {
+ advanceUserVersion(user.id());
+ return UserEntity.builder()
+ .withId(oldUser.id())
+ .withName(oldUser.name())
+ .withNamespace(oldUser.namespace())
+ .withExternalId(oldUser.externalId())
+ .withEnabled(false)
+ .withRoleNames(oldUser.roleNames())
+ .withRoleIds(oldUser.roleIds())
+ .withAuditInfo(oldUser.auditInfo())
+ .build();
+ }));
+
+ assertTrue(queryEnabledByExtId(user.externalId()));
+ }
+
+ @TestTemplate
+ void testDeleteUserById() throws IOException {
+ UserMetaService service = userMetaService();
+ UserEntity user = userWithExtId("delete-by-id-user",
"delete-by-id-ext-id");
+ service.insertUser(user, false);
+
+ assertTrue(service.deleteUserById(metalakeName, user.id()));
+ assertFalse(service.deleteUserById(metalakeName, user.id()));
+ }
+
private UserMetaService userMetaService() throws IOException {
createAndInsertMakeLake(metalakeName);
return UserMetaService.getInstance();
}
+ private MetalakePO getMetalakePO() {
+ return SessionUtils.getWithoutCommit(
+ MetalakeMetaMapper.class, mapper ->
mapper.selectMetalakeMetaByName(metalakeName));
+ }
+
+ private UserPO getUserPO(String userName) {
+ MetalakePO metalakePO = getMetalakePO();
+ return SessionUtils.getWithoutCommit(
+ UserMetaMapper.class,
+ mapper ->
mapper.selectUserMetaByMetalakeIdAndName(metalakePO.getMetalakeId(), userName));
+ }
+
private void assertThrowsExt(Class<? extends Exception> type, Executable
executable) {
Assertions.assertThrows(type, executable);
}
@@ -1577,4 +1816,19 @@ class TestUserMetaService extends TestJDBCBackend {
}
return count;
}
+
+ private void advanceUserVersion(long userId) {
+ try (SqlSession sqlSession =
+
SqlSessionFactoryHelper.getInstance().getSqlSessionFactory().openSession(true);
+ Connection connection = sqlSession.getConnection();
+ Statement statement = connection.createStatement()) {
+ assertEquals(
+ 1,
+ statement.executeUpdate(
+ "UPDATE user_meta SET current_version = current_version + 1
WHERE user_id = "
+ + userId));
+ } catch (SQLException e) {
+ throw new RuntimeException("Advance user 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 e3121b74c5..8e1c563073 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
@@ -61,6 +61,7 @@ import org.apache.gravitino.meta.TableEntity;
import org.apache.gravitino.meta.TableStatisticEntity;
import org.apache.gravitino.meta.TagEntity;
import org.apache.gravitino.meta.TopicEntity;
+import org.apache.gravitino.meta.UserEntity;
import org.apache.gravitino.policy.Policy;
import org.apache.gravitino.policy.PolicyContent;
import org.apache.gravitino.policy.PolicyContents;
@@ -93,6 +94,7 @@ import org.apache.gravitino.storage.relational.po.TablePO;
import org.apache.gravitino.storage.relational.po.TagMetadataObjectRelPO;
import org.apache.gravitino.storage.relational.po.TagPO;
import org.apache.gravitino.storage.relational.po.TopicPO;
+import org.apache.gravitino.storage.relational.po.UserPO;
import org.apache.gravitino.utils.NameIdentifierUtil;
import org.apache.gravitino.utils.NamespaceUtil;
import org.junit.jupiter.api.Assertions;
@@ -841,6 +843,33 @@ public class TestPOConverters {
assertEquals(8, updatePO3.getFilesetVersionPOs().get(0).getVersion());
}
+ @Test
+ public void testUpdateUserPOVersionUsesCurrentVersion() {
+ AuditInfo auditInfo =
+
AuditInfo.builder().withCreator("creator").withCreateTime(FIX_INSTANT).build();
+ UserEntity user =
+
UserEntity.builder().withId(1L).withName("user").withAuditInfo(auditInfo).build();
+ UserPO initialUserPO =
+ POConverters.initializeUserPOWithVersion(user,
UserPO.builder().withMetalakeId(1L));
+ UserPO userPO =
+ UserPO.builder()
+ .withUserId(initialUserPO.getUserId())
+ .withUserName(initialUserPO.getUserName())
+ .withMetalakeId(initialUserPO.getMetalakeId())
+ .withExternalId(initialUserPO.getExternalId())
+ .withEnabled(initialUserPO.getEnabled())
+ .withAuditInfo(initialUserPO.getAuditInfo())
+ .withCurrentVersion(7L)
+ .withLastVersion(3L)
+ .withDeletedAt(initialUserPO.getDeletedAt())
+ .build();
+
+ UserPO updatedUserPO = POConverters.updateUserPOWithVersion(userPO, user);
+
+ assertEquals(8, updatedUserPO.getCurrentVersion());
+ assertEquals(8, updatedUserPO.getLastVersion());
+ }
+
@Test
public void testFromPolicyPO() throws JsonProcessingException {
ImmutableSet<MetadataObject.Type> supportedObjectTypes =
