yuqi1129 commented on code in PR #12781:
URL: https://github.com/apache/gravitino/pull/12781#discussion_r3904178690
##########
core/src/main/java/org/apache/gravitino/storage/relational/service/TagMetaService.java:
##########
@@ -625,6 +678,66 @@ private static void validateAllowedValue(TagPO tagPO,
TagValue tagValue)
Arrays.toString(allowedValues));
}
+ void lockMetalakeForTagCreate(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()));
Review Comment:
Handled by the shared helper: `OccWriteSupport.lockParentForChildWrite`
returns/throws on `currentParent == null` before it ever calls the predicate,
so a metalake that is concurrently deleted produces a `NoSuchEntityException`,
not an NPE. That branch is covered by
`TestOccWriteSupport.testLockParentForChildWriteNotFoundThrows`.
##########
core/src/main/java/org/apache/gravitino/storage/relational/service/TagMetaService.java:
##########
@@ -625,6 +678,66 @@ private static void validateAllowedValue(TagPO tagPO,
TagValue tagValue)
Arrays.toString(allowedValues));
}
+ void lockMetalakeForTagCreate(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 void deleteTagWithVersion(NameIdentifier identifier, TagPO
observedTagPO) {
+ OccWriteSupport.deleteWithVersion(
+ () ->
+ SessionUtils.getWithoutCommit(
+ TagMetaMapper.class,
+ mapper ->
+ mapper.softDeleteTagMetaByIdAndVersion(
+ observedTagPO.getTagId(),
observedTagPO.getCurrentVersion())),
+ () -> tagWriteFailure(identifier, observedTagPO));
+ }
+
+ private RuntimeException tagWriteFailure(NameIdentifier identifier, TagPO
observedTagPO) {
+ return OccWriteSupport.writeFailure(
+ identifier,
+ Entity.EntityType.TAG,
+ () ->
+ SessionUtils.getWithoutCommit(
+ TagMetaMapper.class,
+ mapper ->
mapper.selectTagByTagIdForUpdate(observedTagPO.getTagId())),
+ null,
+ current ->
+ Objects.equals(current.getTagName(), observedTagPO.getTagName())
Review Comment:
Same shape as the other one: `OccWriteSupport.writeFailure` checks
`currentPO == null` first and only then applies the predicate, so a
concurrently deleted tag is classified as `NoSuchEntityException`. Covered by
`TestOccWriteSupport.testWriteFailureReturnsNoSuchEntityWhenNotFound`.
##########
core/src/main/java/org/apache/gravitino/storage/relational/service/TagMetaService.java:
##########
@@ -625,6 +678,66 @@ private static void validateAllowedValue(TagPO tagPO,
TagValue tagValue)
Arrays.toString(allowedValues));
}
+ void lockMetalakeForTagCreate(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 void deleteTagWithVersion(NameIdentifier identifier, TagPO
observedTagPO) {
+ OccWriteSupport.deleteWithVersion(
+ () ->
+ SessionUtils.getWithoutCommit(
+ TagMetaMapper.class,
+ mapper ->
+ mapper.softDeleteTagMetaByIdAndVersion(
+ observedTagPO.getTagId(),
observedTagPO.getCurrentVersion())),
+ () -> tagWriteFailure(identifier, observedTagPO));
+ }
+
+ private RuntimeException tagWriteFailure(NameIdentifier identifier, TagPO
observedTagPO) {
+ return OccWriteSupport.writeFailure(
+ identifier,
+ Entity.EntityType.TAG,
+ () ->
+ SessionUtils.getWithoutCommit(
+ TagMetaMapper.class,
+ mapper ->
mapper.selectTagByTagIdForUpdate(observedTagPO.getTagId())),
+ null,
+ current ->
+ Objects.equals(current.getTagName(), observedTagPO.getTagName())
+ && Objects.equals(current.getMetalakeId(),
observedTagPO.getMetalakeId()));
+ }
+
+ private List<TagPO> lockTagsForAssignment(List<TagPO> observedTagPOs) {
+ List<TagPO> sortedTagPOs = new ArrayList<>(observedTagPOs);
+ sortedTagPOs.sort(Comparator.comparingLong(TagPO::getTagId));
+ List<TagPO> lockedTagPOs = new ArrayList<>(sortedTagPOs.size());
+ for (TagPO observedTagPO : sortedTagPOs) {
+ TagPO lockedTagPO =
+ SessionUtils.getWithoutCommit(
+ TagMetaMapper.class,
+ mapper ->
mapper.selectTagByTagIdForUpdate(observedTagPO.getTagId()));
+ if (lockedTagPO == null
+ || !Objects.equals(lockedTagPO.getTagName(),
observedTagPO.getTagName())
+ || !Objects.equals(lockedTagPO.getMetalakeId(),
observedTagPO.getMetalakeId())) {
+ throw new NoSuchEntityException(
+ NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
+ Entity.EntityType.TAG.name().toLowerCase(),
+ observedTagPO.getTagName());
+ }
+ lockedTagPOs.add(lockedTagPO);
+ }
+ return lockedTagPOs;
+ }
Review Comment:
Fixed. The tags are now locked by one statement,
`listTagPOsByTagIdsForUpdate`, which selects `WHERE tag_id IN (...) ORDER BY
tag_id FOR UPDATE`; the returned rows are then validated against the observed
set. That keeps the tag-ID lock order and collapses the round trips to one.
##########
core/src/test/java/org/apache/gravitino/storage/relational/service/TestTagMetaService.java:
##########
@@ -321,6 +327,124 @@ public void testUpdateTag() throws IOException {
Assertions.assertEquals(tagEntity2, loadedTagEntity1);
}
+ @TestTemplate
+ public void testTagAlterDeleteAndOverwriteUseMonotonicVersion() throws
IOException {
+ createAndInsertMakeLake(METALAKE_NAME);
+ TagMetaService tagMetaService = TagMetaService.getInstance();
Review Comment:
The description does list the commands under "How was this patch tested?";
it has been updated with the suites run for this round (`TestTagMetaService`,
`TestPolicyTagRelService`, `TestTagManager`).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
