This is an automated email from the ASF dual-hosted git repository. yuqi1129 pushed a commit to branch feat/12166-occ-drop-cas in repository https://gitbox.apache.org/repos/asf/gravitino.git
commit ed8e5c673aada702a3b37cc0162c4eed28dc7aea Author: yuqi <[email protected]> AuthorDate: Thu Jul 23 21:33:12 2026 +0800 [#12166] improvement(core): always raise current_version on entity update for OCC Make metalake/catalog/schema/topic/tag/user/group/role bump current_version on every successful update (was frozen at the initial version), so a version CAS on the entity row can detect concurrent writes. Strengthen POConverter tests to pin the post-update version. table already incremented. Part of #12166 (alter/version-CAS). Complex versioned entities (fileset/policy/ view/function), the model schema migration, and the UPDATE WHERE slimming follow. --- .../storage/relational/utils/POConverters.java | 44 ++++++----- .../storage/relational/utils/TestPOConverters.java | 91 ++++++++++++++++++++++ 2 files changed, 115 insertions(+), 20 deletions(-) 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 fcb05a6f46..0327a38cce 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 @@ -137,8 +137,9 @@ public class POConverters { public static MetalakePO updateMetalakePOWithVersion( MetalakePO oldMetalakePO, BaseMetalake newMetalake) { Long lastVersion = oldMetalakePO.getLastVersion(); - // Will set the version to the last version + 1 when having some fields need be multiple version - Long nextVersion = lastVersion; + // Raise the version on every successful update so OCC (version CAS) can detect a concurrent + // write: the racing updater's `WHERE current_version = old` matches 0 rows and loses. + Long nextVersion = lastVersion + 1; try { return MetalakePO.builder() .withMetalakeId(newMetalake.id()) @@ -234,8 +235,9 @@ public class POConverters { public static CatalogPO updateCatalogPOWithVersion( CatalogPO oldCatalogPO, CatalogEntity newCatalog, Long metalakeId) { Long lastVersion = oldCatalogPO.getLastVersion(); - // Will set the version to the last version + 1 when having some fields need be multiple version - Long nextVersion = lastVersion; + // Raise the version on every successful update so OCC (version CAS) can detect a concurrent + // write: the racing updater's `WHERE current_version = old` matches 0 rows and loses. + Long nextVersion = lastVersion + 1; try { return CatalogPO.builder() .withCatalogId(newCatalog.id()) @@ -330,8 +332,9 @@ public class POConverters { */ public static SchemaPO updateSchemaPOWithVersion(SchemaPO oldSchemaPO, SchemaEntity newSchema) { Long lastVersion = oldSchemaPO.getLastVersion(); - // Will set the version to the last version + 1 when having some fields need be multiple version - Long nextVersion = lastVersion; + // Raise the version on every successful update so OCC (version CAS) can detect a concurrent + // write: the racing updater's `WHERE current_version = old` matches 0 rows and loses. + Long nextVersion = lastVersion + 1; try { return SchemaPO.builder() .withSchemaId(oldSchemaPO.getSchemaId()) @@ -922,8 +925,9 @@ public class POConverters { public static TopicPO updateTopicPOWithVersion(TopicPO oldTopicPO, TopicEntity newEntity) { Long lastVersion = oldTopicPO.getLastVersion(); - // Will set the version to the last version + 1 when having some fields need be multiple version - Long nextVersion = lastVersion; + // Raise the version on every successful update so OCC (version CAS) can detect a concurrent + // write: the racing updater's `WHERE current_version = old` matches 0 rows and loses. + Long nextVersion = lastVersion + 1; try { return TopicPO.builder() .withTopicId(oldTopicPO.getTopicId()) @@ -977,9 +981,9 @@ public class POConverters { */ 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; + // Raise the version on every successful update so OCC (version CAS) can detect a concurrent + // write: the racing updater's `WHERE current_version = old` matches 0 rows and loses. + Long nextVersion = lastVersion + 1; try { return UserPO.builder() .withUserId(oldUserPO.getUserId()) @@ -1259,9 +1263,9 @@ public class POConverters { */ 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; + // Raise the version on every successful update so OCC (version CAS) can detect a concurrent + // write: the racing updater's `WHERE current_version = old` matches 0 rows and loses. + Long nextVersion = lastVersion + 1; try { return GroupPO.builder() .withGroupId(oldGroupPO.getGroupId()) @@ -1387,9 +1391,9 @@ public class POConverters { 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; + // Raise the version on every successful update so OCC (version CAS) can detect a concurrent + // write: the racing updater's `WHERE current_version = old` matches 0 rows and loses. + Long nextVersion = lastVersion + 1; try { return RolePO.builder() .withRoleId(oldRolePO.getRoleId()) @@ -1445,9 +1449,9 @@ public class POConverters { public static TagPO updateTagPOWithVersion(TagPO oldTagPO, TagEntity newEntity) { Long lastVersion = oldTagPO.getLastVersion(); - // TODO: set the version to the last version + 1 when having some fields need be multiple - // version - Long nextVersion = lastVersion; + // Raise the version on every successful update so OCC (version CAS) can detect a concurrent + // write: the racing updater's `WHERE current_version = old` matches 0 rows and loses. + Long nextVersion = lastVersion + 1; try { return TagPO.builder() .withTagId(oldTagPO.getTagId()) 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 bff96b2bd6..4999aa2702 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,9 +51,11 @@ 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; +import org.apache.gravitino.meta.RoleEntity; import org.apache.gravitino.meta.SchemaEntity; import org.apache.gravitino.meta.SchemaVersion; import org.apache.gravitino.meta.StatisticEntity; @@ -61,6 +63,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; @@ -79,6 +82,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; @@ -86,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; @@ -93,6 +98,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; @@ -666,6 +672,9 @@ public class TestPOConverters { assertEquals(1, initPO.getLastVersion()); assertEquals(0, initPO.getDeletedAt()); assertEquals("this is test2", updatePO.getMetalakeComment()); + // A successful update must raise the version so OCC (version CAS) can detect concurrent writes. + assertEquals(2, updatePO.getCurrentVersion()); + assertEquals(2, updatePO.getLastVersion()); } @Test @@ -680,6 +689,9 @@ public class TestPOConverters { assertEquals(1, initPO.getLastVersion()); assertEquals(0, initPO.getDeletedAt()); assertEquals("this is test2", updatePO.getCatalogComment()); + // A successful update must raise the version so OCC (version CAS) can detect concurrent writes. + assertEquals(2, updatePO.getCurrentVersion()); + assertEquals(2, updatePO.getLastVersion()); } @Test @@ -697,6 +709,9 @@ public class TestPOConverters { assertEquals(1, initPO.getLastVersion()); assertEquals(0, initPO.getDeletedAt()); assertEquals("this is test2", updatePO.getSchemaComment()); + // A successful update must raise the version so OCC (version CAS) can detect concurrent writes. + assertEquals(2, updatePO.getCurrentVersion()); + assertEquals(2, updatePO.getLastVersion()); } @Test @@ -717,6 +732,79 @@ public class TestPOConverters { assertEquals("test", updatePO.getTableName()); } + @Test + public void testUpdateTopicPOVersionIncrements() throws JsonProcessingException { + TopicPO initPO = + createTopicPO(1L, "test", 1L, 1L, 1L, "this is test", ImmutableMap.of("key", "value")); + TopicEntity updatedTopic = + createTopic( + 1L, + "test", + NamespaceUtil.ofTopic("test_metalake", "test_catalog", "test_schema"), + "this is test2", + ImmutableMap.of("key", "value")); + + TopicPO updatePO = POConverters.updateTopicPOWithVersion(initPO, updatedTopic); + + // A successful update must raise the version so OCC (version CAS) can detect concurrent writes. + assertEquals(2, updatePO.getCurrentVersion()); + assertEquals(2, updatePO.getLastVersion()); + } + + @Test + public void testUpdateUserPOVersionIncrements() throws JsonProcessingException { + AuditInfo auditInfo = + AuditInfo.builder().withCreator("creator").withCreateTime(FIX_INSTANT).build(); + UserEntity user = + UserEntity.builder().withId(1L).withName("test").withAuditInfo(auditInfo).build(); + UserEntity updatedUser = + UserEntity.builder().withId(1L).withName("test2").withAuditInfo(auditInfo).build(); + UserPO initPO = + POConverters.initializeUserPOWithVersion(user, UserPO.builder().withMetalakeId(1L)); + + UserPO updatePO = POConverters.updateUserPOWithVersion(initPO, updatedUser); + + // A successful update must raise the version so OCC (version CAS) can detect concurrent writes. + assertEquals(2, updatePO.getCurrentVersion()); + assertEquals(2, updatePO.getLastVersion()); + } + + @Test + public void testUpdateGroupPOVersionIncrements() throws JsonProcessingException { + AuditInfo auditInfo = + AuditInfo.builder().withCreator("creator").withCreateTime(FIX_INSTANT).build(); + GroupEntity group = + GroupEntity.builder().withId(1L).withName("test").withAuditInfo(auditInfo).build(); + GroupEntity updatedGroup = + GroupEntity.builder().withId(1L).withName("test2").withAuditInfo(auditInfo).build(); + GroupPO initPO = + POConverters.initializeGroupPOWithVersion(group, GroupPO.builder().withMetalakeId(1L)); + + GroupPO updatePO = POConverters.updateGroupPOWithVersion(initPO, updatedGroup); + + // A successful update must raise the version so OCC (version CAS) can detect concurrent writes. + assertEquals(2, updatePO.getCurrentVersion()); + assertEquals(2, updatePO.getLastVersion()); + } + + @Test + public void testUpdateRolePOVersionIncrements() throws JsonProcessingException { + AuditInfo auditInfo = + AuditInfo.builder().withCreator("creator").withCreateTime(FIX_INSTANT).build(); + RoleEntity role = + RoleEntity.builder().withId(1L).withName("test").withAuditInfo(auditInfo).build(); + RoleEntity updatedRole = + RoleEntity.builder().withId(1L).withName("test2").withAuditInfo(auditInfo).build(); + RolePO initPO = + POConverters.initializeRolePOWithVersion(role, RolePO.builder().withMetalakeId(1L)); + + RolePO updatePO = POConverters.updateRolePOWithVersion(initPO, updatedRole); + + // A successful update must raise the version so OCC (version CAS) can detect concurrent writes. + assertEquals(2, updatePO.getCurrentVersion()); + assertEquals(2, updatePO.getLastVersion()); + } + @Test public void testUpdateFilesetPOVersion() throws JsonProcessingException { Map<String, String> properties = new HashMap<>(); @@ -880,6 +968,9 @@ public class TestPOConverters { assertEquals(1, initPO.getLastVersion()); assertEquals(0, initPO.getDeletedAt()); assertEquals("this is test2", updatePO.getComment()); + // A successful update must raise the version so OCC (version CAS) can detect concurrent writes. + assertEquals(2, updatePO.getCurrentVersion()); + assertEquals(2, updatePO.getLastVersion()); } @Test
