This is an automated email from the ASF dual-hosted git repository.
jerryshao 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 d326c6ac2a [#12775] feat(core): Add OCC for tag metadata (#12781)
d326c6ac2a is described below
commit d326c6ac2ae3de7107018c17f0eca04e10dabd9c
Author: Qi Yu <[email protected]>
AuthorDate: Mon Sep 7 14:02:01 2026 +0800
[#12775] feat(core): Add OCC for tag metadata (#12781)
### What changes were proposed in this pull request?
- Apply stable-ID and expected-version CAS to tag alter and delete
operations.
- Replace database-specific tag overwrite upserts with locked, monotonic
root updates that preserve the stored tag ID.
- Lock parent metalakes during creation and tag roots during assignment
changes.
- Keep tag deletion and all dependent relationship cleanup, including
policy-on-tag rows, atomic.
- Add real two-transaction interleaving coverage for create/delete,
overwrite/rename, and assignment/rename races.
### Why are the changes needed?
Concurrent tag writes could otherwise cause lost updates, stale
deletions, ID/version resets, or partial relationship cleanup.
Fix: #12775
### Does this PR introduce _any_ user-facing change?
No API or configuration change. Concurrent stale writes now fail
deterministically; a tag deleted or renamed by a competing transaction
preserves the existing idempotent delete result.
### How was this patch tested?
- `./gradlew :core:spotlessApply :core:compileTestJava :core:javadoc`
- `./gradlew :core:test --tests
org.apache.gravitino.storage.relational.service.TestTagMetaService
--tests
org.apache.gravitino.storage.relational.service.TestPolicyTagRelService
--tests org.apache.gravitino.tag.TestTagManager -PskipDockerTests=false`
- `TestTagMetaService` ran against H2, MySQL, and PostgreSQL.
---
.../relational/mapper/PolicyTagRelMapper.java | 9 +
.../mapper/PolicyTagRelSQLProviderFactory.java | 5 +
.../storage/relational/mapper/TagMetaMapper.java | 40 +-
.../mapper/TagMetaSQLProviderFactory.java | 22 +-
.../mapper/TagMetadataObjectRelMapper.java | 11 +
.../TagMetadataObjectRelSQLProviderFactory.java | 5 +
.../provider/base/PolicyTagRelBaseSQLProvider.java | 9 +
.../provider/base/TagMetaBaseSQLProvider.java | 89 ++-
.../base/TagMetadataObjectRelBaseSQLProvider.java | 9 +
.../postgresql/TagMetaPostgreSQLProvider.java | 50 +-
.../TagMetadataObjectRelPostgreSQLProvider.java | 8 +
.../relational/service/PolicyTagRelService.java | 26 +-
.../storage/relational/service/TagMetaService.java | 370 ++++++++++---
.../storage/relational/utils/POConverters.java | 5 +-
.../java/org/apache/gravitino/tag/TagManager.java | 18 +
.../relational/service/TestTagMetaService.java | 597 +++++++++++++++++++++
16 files changed, 1082 insertions(+), 191 deletions(-)
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyTagRelMapper.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyTagRelMapper.java
index 370c2b2f5e..35a97b04a8 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyTagRelMapper.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyTagRelMapper.java
@@ -91,6 +91,15 @@ public interface PolicyTagRelMapper {
@UpdateProvider(type = PolicyTagRelSQLProviderFactory.class, method =
"softDeleteByMetalakeId")
int softDeleteByMetalakeId(@Param("metalakeId") Long metalakeId);
+ /**
+ * Soft-deletes every active policy relation for a tag.
+ *
+ * @param tagId The tag ID.
+ * @return The number of affected rows.
+ */
+ @UpdateProvider(type = PolicyTagRelSQLProviderFactory.class, method =
"softDeleteByTagId")
+ int softDeleteByTagId(@Param("tagId") Long tagId);
+
/**
* Physically deletes expired relation rows.
*
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyTagRelSQLProviderFactory.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyTagRelSQLProviderFactory.java
index d9cdb827ce..5a82d68d77 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyTagRelSQLProviderFactory.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyTagRelSQLProviderFactory.java
@@ -82,6 +82,11 @@ public class PolicyTagRelSQLProviderFactory {
return getProvider().softDeleteByMetalakeId(metalakeId);
}
+ /** Delegates tag deletion cleanup. */
+ public static String softDeleteByTagId(@Param("tagId") Long tagId) {
+ return getProvider().softDeleteByTagId(tagId);
+ }
+
/** Delegates expired relation cleanup. */
public static String deleteByLegacyTimeline(
@Param("legacyTimeline") Long legacyTimeline, @Param("limit") int limit)
{
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TagMetaMapper.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TagMetaMapper.java
index 2d6095bf03..721499826f 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TagMetaMapper.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TagMetaMapper.java
@@ -53,22 +53,37 @@ public interface TagMetaMapper {
TagPO selectTagMetaByMetalakeIdAndName(
@Param("metalakeId") Long metalakeId, @Param("name") String tagName);
+ /**
+ * Selects and exclusively locks an active tag by its natural key.
+ *
+ * @param metalakeId The metalake ID.
+ * @param tagName The tag name.
+ * @return The locked tag, or null if the natural key is not active.
+ */
+ @SelectProvider(
+ type = TagMetaSQLProviderFactory.class,
+ method = "selectTagMetaByMetalakeIdAndNameForUpdate")
+ TagPO selectTagMetaByMetalakeIdAndNameForUpdate(
+ @Param("metalakeId") Long metalakeId, @Param("name") String tagName);
+
@InsertProvider(type = TagMetaSQLProviderFactory.class, method =
"insertTagMeta")
void insertTagMeta(@Param("tagMeta") TagPO tagPO);
- @InsertProvider(
- type = TagMetaSQLProviderFactory.class,
- method = "insertTagMetaOnDuplicateKeyUpdate")
- void insertTagMetaOnDuplicateKeyUpdate(@Param("tagMeta") TagPO tagPO);
-
@UpdateProvider(type = TagMetaSQLProviderFactory.class, method =
"updateTagMeta")
Integer updateTagMeta(@Param("newTagMeta") TagPO newTagPO,
@Param("oldTagMeta") TagPO oldTagPO);
+ /**
+ * Soft-deletes an active tag when its OCC version still matches.
+ *
+ * @param tagId The tag ID.
+ * @param currentVersion The version observed by the caller.
+ * @return The number of affected rows.
+ */
@UpdateProvider(
type = TagMetaSQLProviderFactory.class,
- method = "softDeleteTagMetaByMetalakeAndTagName")
- Integer softDeleteTagMetaByMetalakeAndTagName(
- @Param("metalakeName") String metalakeName, @Param("tagName") String
tagName);
+ method = "softDeleteTagMetaByIdAndVersion")
+ Integer softDeleteTagMetaByIdAndVersion(
+ @Param("tagId") Long tagId, @Param("currentVersion") Long
currentVersion);
@UpdateProvider(type = TagMetaSQLProviderFactory.class, method =
"softDeleteTagMetasByMetalakeId")
void softDeleteTagMetasByMetalakeId(@Param("metalakeId") Long metalakeId);
@@ -92,6 +107,15 @@ public interface TagMetaMapper {
@SelectProvider(type = TagMetaSQLProviderFactory.class, method =
"listTagPOsByTagIds")
List<TagPO> listTagPOsByTagIds(@Param("tagIds") List<Long> tagIds);
+ /**
+ * Selects and exclusively locks the active tags with the given IDs, in
ascending ID order.
+ *
+ * @param tagIds The tag IDs to lock.
+ * @return The locked tags. Tags that are not active are absent from the
result.
+ */
+ @SelectProvider(type = TagMetaSQLProviderFactory.class, method =
"listTagPOsByTagIdsForUpdate")
+ List<TagPO> listTagPOsByTagIdsForUpdate(@Param("tagIds") List<Long> tagIds);
+
@SelectProvider(type = TagMetaSQLProviderFactory.class, method =
"batchSelectTagByIdentifier")
List<TagPO> batchSelectTagByIdentifier(
@Param("metalakeName") String metalakeName, @Param("tagNames")
List<String> tagNames);
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TagMetaSQLProviderFactory.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TagMetaSQLProviderFactory.java
index 400bcc0e08..b7feb9b4b7 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TagMetaSQLProviderFactory.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TagMetaSQLProviderFactory.java
@@ -74,18 +74,15 @@ public class TagMetaSQLProviderFactory {
return getProvider().insertTagMeta(tagPO);
}
- public static String insertTagMetaOnDuplicateKeyUpdate(@Param("tagMeta")
TagPO tagPO) {
- return getProvider().insertTagMetaOnDuplicateKeyUpdate(tagPO);
- }
-
public static String updateTagMeta(
@Param("newTagMeta") TagPO newTagPO, @Param("oldTagMeta") TagPO
oldTagPO) {
return getProvider().updateTagMeta(newTagPO, oldTagPO);
}
- public static String softDeleteTagMetaByMetalakeAndTagName(
- @Param("metalakeName") String metalakeName, @Param("tagName") String
tagName) {
- return getProvider().softDeleteTagMetaByMetalakeAndTagName(metalakeName,
tagName);
+ /** Delegates a version-checked tag soft delete. */
+ public static String softDeleteTagMetaByIdAndVersion(
+ @Param("tagId") Long tagId, @Param("currentVersion") Long
currentVersion) {
+ return getProvider().softDeleteTagMetaByIdAndVersion(tagId,
currentVersion);
}
public static String softDeleteTagMetasByMetalakeId(@Param("metalakeId")
Long metalakeId) {
@@ -102,6 +99,12 @@ public class TagMetaSQLProviderFactory {
return getProvider().selectTagMetaByMetalakeIdAndName(metalakeId, name);
}
+ /** Delegates an exclusive-lock tag query by natural key. */
+ public static String selectTagMetaByMetalakeIdAndNameForUpdate(
+ @Param("metalakeId") Long metalakeId, @Param("name") String name) {
+ return getProvider().selectTagMetaByMetalakeIdAndNameForUpdate(metalakeId,
name);
+ }
+
public static String selectTagByTagId(@Param("tagId") Long tagId) {
return getProvider().selectTagByTagId(tagId);
}
@@ -111,6 +114,11 @@ public class TagMetaSQLProviderFactory {
return getProvider().selectTagByTagIdForUpdate(tagId);
}
+ /** Delegates a locking read of several tags. */
+ public static String listTagPOsByTagIdsForUpdate(@Param("tagIds") List<Long>
tagIds) {
+ return getProvider().listTagPOsByTagIdsForUpdate(tagIds);
+ }
+
public static String listTagPOsByTagIds(@Param("tagIds") List<Long> tagIds) {
return getProvider().listTagPOsByTagIds(tagIds);
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TagMetadataObjectRelMapper.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TagMetadataObjectRelMapper.java
index bdb461a3f6..48ecf2636f 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TagMetadataObjectRelMapper.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TagMetadataObjectRelMapper.java
@@ -86,6 +86,17 @@ public interface TagMetadataObjectRelMapper {
Integer softDeleteTagMetadataObjectRelsByMetalakeAndTagName(
@Param("metalakeName") String metalakeName, @Param("tagName") String
tagName);
+ /**
+ * Soft-deletes all active metadata-object assignments for a tag.
+ *
+ * @param tagId The tag ID.
+ * @return The number of affected rows.
+ */
+ @UpdateProvider(
+ type = TagMetadataObjectRelSQLProviderFactory.class,
+ method = "softDeleteTagMetadataObjectRelsByTagId")
+ Integer softDeleteTagMetadataObjectRelsByTagId(@Param("tagId") Long tagId);
+
@UpdateProvider(
type = TagMetadataObjectRelSQLProviderFactory.class,
method = "softDeleteTagMetadataObjectRelsByMetalakeId")
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TagMetadataObjectRelSQLProviderFactory.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TagMetadataObjectRelSQLProviderFactory.java
index 43ec136a93..f604d2aac6 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TagMetadataObjectRelSQLProviderFactory.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TagMetadataObjectRelSQLProviderFactory.java
@@ -107,6 +107,11 @@ public class TagMetadataObjectRelSQLProviderFactory {
return
getProvider().softDeleteTagMetadataObjectRelsByMetalakeAndTagName(metalakeName,
tagName);
}
+ /** Delegates cleanup of metadata-object assignments by tag ID. */
+ public static String softDeleteTagMetadataObjectRelsByTagId(@Param("tagId")
Long tagId) {
+ return getProvider().softDeleteTagMetadataObjectRelsByTagId(tagId);
+ }
+
public static String softDeleteTagMetadataObjectRelsByMetalakeId(
@Param("metalakeId") Long metalakeId) {
return
getProvider().softDeleteTagMetadataObjectRelsByMetalakeId(metalakeId);
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/PolicyTagRelBaseSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/PolicyTagRelBaseSQLProvider.java
index b26a842da4..826e37ba88 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/PolicyTagRelBaseSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/PolicyTagRelBaseSQLProvider.java
@@ -90,6 +90,15 @@ public class PolicyTagRelBaseSQLProvider {
+ " AND deleted_at = 0";
}
+ /** Returns SQL for soft-deleting policy relations when a tag is deleted. */
+ public String softDeleteByTagId(@Param("tagId") Long tagId) {
+ return "UPDATE "
+ + POLICY_TAG_RELATION_TABLE_NAME
+ + " SET deleted_at = "
+ + deletedAtNowExpression()
+ + " WHERE tag_id = #{tagId} AND deleted_at = 0";
+ }
+
/** Returns SQL for physically deleting expired relation rows. */
public String deleteByLegacyTimeline(
@Param("legacyTimeline") Long legacyTimeline, @Param("limit") int limit)
{
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/TagMetaBaseSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/TagMetaBaseSQLProvider.java
index 65a2b4bcda..1b13ad6e1f 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/TagMetaBaseSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/TagMetaBaseSQLProvider.java
@@ -123,36 +123,6 @@ public class TagMetaBaseSQLProvider {
+ " )";
}
- public String insertTagMetaOnDuplicateKeyUpdate(@Param("tagMeta") TagPO
tagPO) {
- return "INSERT INTO "
- + TAG_TABLE_NAME
- + " (tag_id, tag_name,"
- + " metalake_id, tag_comment, properties, allowed_values, audit_info,"
- + " current_version, last_version, deleted_at)"
- + " VALUES ("
- + " #{tagMeta.tagId},"
- + " #{tagMeta.tagName},"
- + " #{tagMeta.metalakeId},"
- + " #{tagMeta.comment},"
- + " #{tagMeta.properties},"
- + " #{tagMeta.allowedValues},"
- + " #{tagMeta.auditInfo},"
- + " #{tagMeta.currentVersion},"
- + " #{tagMeta.lastVersion},"
- + " #{tagMeta.deletedAt}"
- + " )"
- + " ON DUPLICATE KEY UPDATE"
- + " tag_name = #{tagMeta.tagName},"
- + " metalake_id = #{tagMeta.metalakeId},"
- + " tag_comment = #{tagMeta.comment},"
- + " properties = #{tagMeta.properties},"
- + " allowed_values = #{tagMeta.allowedValues},"
- + " audit_info = #{tagMeta.auditInfo},"
- + " current_version = #{tagMeta.currentVersion},"
- + " last_version = #{tagMeta.lastVersion},"
- + " deleted_at = #{tagMeta.deletedAt}";
- }
-
public String updateTagMeta(
@Param("newTagMeta") TagPO newTagPO, @Param("oldTagMeta") TagPO
oldTagPO) {
return "UPDATE "
@@ -166,28 +136,19 @@ public class TagMetaBaseSQLProvider {
+ " last_version = #{newTagMeta.lastVersion},"
+ " deleted_at = #{newTagMeta.deletedAt}"
+ " WHERE tag_id = #{oldTagMeta.tagId}"
- + " AND metalake_id = #{oldTagMeta.metalakeId}"
- + " AND tag_name = #{oldTagMeta.tagName}"
- + " AND (tag_comment = #{oldTagMeta.comment}"
- + " OR (tag_comment IS NULL and #{oldTagMeta.comment} IS NULL))"
- + " AND properties = #{oldTagMeta.properties}"
- + " AND audit_info = #{oldTagMeta.auditInfo}"
+ " AND current_version = #{oldTagMeta.currentVersion}"
- + " AND last_version = #{oldTagMeta.lastVersion}"
+ " AND deleted_at = 0";
}
- public String softDeleteTagMetaByMetalakeAndTagName(
- @Param("metalakeName") String metalakeName, @Param("tagName") String
tagName) {
+ /** Returns SQL that soft-deletes a tag using its stable ID and observed OCC
version. */
+ public String softDeleteTagMetaByIdAndVersion(
+ @Param("tagId") Long tagId, @Param("currentVersion") Long
currentVersion) {
return "UPDATE "
+ TAG_TABLE_NAME
- + " tm SET tm.deleted_at = "
+ + " SET deleted_at = "
+ DatabaseTimeSQL.MYSQL
- + " WHERE tm.metalake_id IN ("
- + " SELECT mm.metalake_id FROM "
- + MetalakeMetaMapper.TABLE_NAME
- + " mm WHERE mm.metalake_name = #{metalakeName} AND mm.deleted_at = 0)"
- + " AND tm.tag_name = #{tagName} AND tm.deleted_at = 0";
+ + " WHERE tag_id = #{tagId} AND current_version = #{currentVersion}"
+ + " AND deleted_at = 0";
}
public String softDeleteTagMetasByMetalakeId(@Param("metalakeId") Long
metalakeId) {
@@ -211,6 +172,7 @@ public class TagMetaBaseSQLProvider {
+ " metalake_id as metalakeId,"
+ " tag_comment as comment,"
+ " properties as properties,"
+ + " allowed_values as allowedValues,"
+ " audit_info as auditInfo,"
+ " current_version as currentVersion,"
+ " last_version as lastVersion,"
@@ -220,11 +182,18 @@ public class TagMetaBaseSQLProvider {
+ " WHERE metalake_id = #{metalakeId} AND tag_name = #{name} and
deleted_at = 0";
}
+ /** Returns SQL that selects and exclusively locks an active tag by its
natural key. */
+ public String selectTagMetaByMetalakeIdAndNameForUpdate(
+ @Param("metalakeId") Long metalakeId, @Param("name") String name) {
+ return selectTagMetaByMetalakeIdAndName(metalakeId, name) + " FOR UPDATE";
+ }
+
public String selectTagByTagId(@Param("tagId") Long tagId) {
return "SELECT tag_id as tagId, tag_name as tagName,"
+ " metalake_id as metalakeId,"
+ " tag_comment as comment,"
+ " properties as properties,"
+ + " allowed_values as allowedValues,"
+ " audit_info as auditInfo,"
+ " current_version as currentVersion,"
+ " last_version as lastVersion,"
@@ -245,6 +214,7 @@ public class TagMetaBaseSQLProvider {
+ " metalake_id as metalakeId,"
+ " tag_comment as comment,"
+ " properties as properties,"
+ + " allowed_values as allowedValues,"
+ " audit_info as auditInfo,"
+ " current_version as currentVersion,"
+ " last_version as lastVersion,"
@@ -260,6 +230,34 @@ public class TagMetaBaseSQLProvider {
+ "</script>";
}
+ /**
+ * Returns SQL that selects and exclusively locks several active tags,
ordered by tag ID so that
+ * concurrent callers take the row locks in the same order.
+ */
+ public String listTagPOsByTagIdsForUpdate(@Param("tagIds") List<Long>
tagIds) {
+ return "<script>"
+ + "SELECT tag_id as tagId, tag_name as tagName,"
+ + " metalake_id as metalakeId,"
+ + " tag_comment as comment,"
+ + " properties as properties,"
+ + " allowed_values as allowedValues,"
+ + " audit_info as auditInfo,"
+ + " current_version as currentVersion,"
+ + " last_version as lastVersion,"
+ + " deleted_at as deletedAt"
+ + " FROM "
+ + TAG_TABLE_NAME
+ + " WHERE deleted_at = 0"
+ + " AND tag_id IN ("
+ + "<foreach collection='tagIds' item='tagId' separator=','>"
+ + "#{tagId}"
+ + "</foreach>"
+ + ")"
+ + " ORDER BY tag_id"
+ + " FOR UPDATE"
+ + "</script>";
+ }
+
public String batchSelectTagByIdentifier(
@Param("metalakeName") String metalakeName, @Param("tagNames")
List<String> tagNames) {
return "<script>"
@@ -267,6 +265,7 @@ public class TagMetaBaseSQLProvider {
+ " tm.metalake_id as metalakeId,"
+ " tm.tag_comment as comment,"
+ " tm.properties as properties,"
+ + " tm.allowed_values as allowedValues,"
+ " tm.audit_info as auditInfo,"
+ " tm.current_version as currentVersion,"
+ " tm.last_version as lastVersion,"
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/TagMetadataObjectRelBaseSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/TagMetadataObjectRelBaseSQLProvider.java
index b3b1b8d6e9..100b7f0694 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/TagMetadataObjectRelBaseSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/TagMetadataObjectRelBaseSQLProvider.java
@@ -191,6 +191,15 @@ public class TagMetadataObjectRelBaseSQLProvider {
+ " AND tm.tag_name = #{tagName} AND tm.deleted_at = 0) AND
te.deleted_at = 0";
}
+ /** Returns SQL that soft-deletes every active metadata-object assignment
for a tag ID. */
+ public String softDeleteTagMetadataObjectRelsByTagId(@Param("tagId") Long
tagId) {
+ return "UPDATE "
+ + TagMetadataObjectRelMapper.TAG_METADATA_OBJECT_RELATION_TABLE_NAME
+ + " SET deleted_at = (UNIX_TIMESTAMP() * 1000.0)"
+ + " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000"
+ + " WHERE tag_id = #{tagId} AND deleted_at = 0";
+ }
+
public String
softDeleteTagMetadataObjectRelsByMetalakeId(@Param("metalakeId") Long
metalakeId) {
return "UPDATE "
+ TagMetadataObjectRelMapper.TAG_METADATA_OBJECT_RELATION_TABLE_NAME
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/TagMetaPostgreSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/TagMetaPostgreSQLProvider.java
index 422f6aee73..8b10154454 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/TagMetaPostgreSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/TagMetaPostgreSQLProvider.java
@@ -20,7 +20,6 @@ package
org.apache.gravitino.storage.relational.mapper.provider.postgresql;
import static
org.apache.gravitino.storage.relational.mapper.TagMetaMapper.TAG_TABLE_NAME;
-import org.apache.gravitino.storage.relational.mapper.MetalakeMetaMapper;
import org.apache.gravitino.storage.relational.mapper.provider.DatabaseTimeSQL;
import
org.apache.gravitino.storage.relational.mapper.provider.base.TagMetaBaseSQLProvider;
import org.apache.gravitino.storage.relational.po.TagPO;
@@ -28,16 +27,13 @@ import org.apache.ibatis.annotations.Param;
public class TagMetaPostgreSQLProvider extends TagMetaBaseSQLProvider {
@Override
- public String softDeleteTagMetaByMetalakeAndTagName(String metalakeName,
String tagName) {
+ public String softDeleteTagMetaByIdAndVersion(Long tagId, Long
currentVersion) {
return "UPDATE "
+ TAG_TABLE_NAME
- + " tm SET deleted_at = "
+ + " SET deleted_at = "
+ DatabaseTimeSQL.POSTGRESQL
- + " WHERE tm.metalake_id IN ("
- + " SELECT mm.metalake_id FROM "
- + MetalakeMetaMapper.TABLE_NAME
- + " mm WHERE mm.metalake_name = #{metalakeName} AND mm.deleted_at = 0)"
- + " AND tm.tag_name = #{tagName} AND tm.deleted_at = 0";
+ + " WHERE tag_id = #{tagId} AND current_version = #{currentVersion}"
+ + " AND deleted_at = 0";
}
@Override
@@ -49,37 +45,6 @@ public class TagMetaPostgreSQLProvider extends
TagMetaBaseSQLProvider {
+ " WHERE metalake_id = #{metalakeId} AND deleted_at = 0";
}
- @Override
- public String insertTagMetaOnDuplicateKeyUpdate(TagPO tagPO) {
- return "INSERT INTO "
- + TAG_TABLE_NAME
- + " (tag_id, tag_name,"
- + " metalake_id, tag_comment, properties, allowed_values, audit_info,"
- + " current_version, last_version, deleted_at)"
- + " VALUES ("
- + " #{tagMeta.tagId},"
- + " #{tagMeta.tagName},"
- + " #{tagMeta.metalakeId},"
- + " #{tagMeta.comment},"
- + " #{tagMeta.properties},"
- + " #{tagMeta.allowedValues},"
- + " #{tagMeta.auditInfo},"
- + " #{tagMeta.currentVersion},"
- + " #{tagMeta.lastVersion},"
- + " #{tagMeta.deletedAt}"
- + " )"
- + " ON CONFLICT(tag_id) DO UPDATE SET"
- + " tag_name = #{tagMeta.tagName},"
- + " metalake_id = #{tagMeta.metalakeId},"
- + " tag_comment = #{tagMeta.comment},"
- + " properties = #{tagMeta.properties},"
- + " allowed_values = #{tagMeta.allowedValues},"
- + " audit_info = #{tagMeta.auditInfo},"
- + " current_version = #{tagMeta.currentVersion},"
- + " last_version = #{tagMeta.lastVersion},"
- + " deleted_at = #{tagMeta.deletedAt}";
- }
-
@Override
public String updateTagMeta(
@Param("newTagMeta") TagPO newTagPO, @Param("oldTagMeta") TagPO
oldTagPO) {
@@ -94,14 +59,7 @@ public class TagMetaPostgreSQLProvider extends
TagMetaBaseSQLProvider {
+ " last_version = #{newTagMeta.lastVersion},"
+ " deleted_at = #{newTagMeta.deletedAt}"
+ " WHERE tag_id = #{oldTagMeta.tagId}"
- + " AND metalake_id = #{oldTagMeta.metalakeId}"
- + " AND tag_name = #{oldTagMeta.tagName}"
- + " AND (tag_comment = #{oldTagMeta.comment} "
- + " OR (CAST(tag_comment AS VARCHAR) IS NULL AND
CAST(#{oldTagMeta.comment} AS VARCHAR) IS NULL))"
- + " AND properties = #{oldTagMeta.properties}"
- + " AND audit_info = #{oldTagMeta.auditInfo}"
+ " AND current_version = #{oldTagMeta.currentVersion}"
- + " AND last_version = #{oldTagMeta.lastVersion}"
+ " AND deleted_at = 0";
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/TagMetadataObjectRelPostgreSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/TagMetadataObjectRelPostgreSQLProvider.java
index 3fa974c44f..c1292e8dff 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/TagMetadataObjectRelPostgreSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/TagMetadataObjectRelPostgreSQLProvider.java
@@ -39,6 +39,14 @@ import
org.apache.gravitino.storage.relational.po.TagMetadataObjectRelPO;
import org.apache.ibatis.annotations.Param;
public class TagMetadataObjectRelPostgreSQLProvider extends
TagMetadataObjectRelBaseSQLProvider {
+ @Override
+ public String softDeleteTagMetadataObjectRelsByTagId(Long tagId) {
+ return "UPDATE "
+ + TAG_METADATA_OBJECT_RELATION_TABLE_NAME
+ + " SET deleted_at = CAST(EXTRACT(EPOCH FROM CURRENT_TIMESTAMP) * 1000
AS BIGINT)"
+ + " WHERE tag_id = #{tagId} AND deleted_at = 0";
+ }
+
@Override
public String
batchDeleteTagMetadataObjectRelsByTagIdsAndValuesAndMetadataObject(
Long metadataObjectId, String metadataObjectType,
List<TagMetadataObjectRelPO> tagRelPOs) {
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/PolicyTagRelService.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/PolicyTagRelService.java
index fc011456c2..49ae66a8d9 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/service/PolicyTagRelService.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/PolicyTagRelService.java
@@ -28,7 +28,6 @@ import java.util.Collections;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
-import java.util.Objects;
import java.util.Set;
import java.util.stream.Collectors;
import org.apache.gravitino.Entity;
@@ -338,21 +337,16 @@ public class PolicyTagRelService {
private static TagPO lockTag(NameIdentifier tagIdentifier) {
String metalake = tagIdentifier.namespace().level(0);
- return SessionUtils.getWithoutCommit(
- TagMetaMapper.class,
- mapper -> {
- TagPO observed = mapper.selectTagMetaByMetalakeAndName(metalake,
tagIdentifier.name());
- if (observed == null) {
- throw noSuchEntity(Entity.EntityType.TAG, tagIdentifier.name());
- }
- TagPO locked = mapper.selectTagByTagIdForUpdate(observed.getTagId());
- if (locked == null
- || !Objects.equals(locked.getTagName(), tagIdentifier.name())
- || !Objects.equals(locked.getMetalakeId(),
observed.getMetalakeId())) {
- throw noSuchEntity(Entity.EntityType.TAG, tagIdentifier.name());
- }
- return locked;
- });
+ TagPO observed =
+ SessionUtils.getWithoutCommit(
+ TagMetaMapper.class,
+ mapper -> mapper.selectTagMetaByMetalakeAndName(metalake,
tagIdentifier.name()));
+ if (observed == null) {
+ throw noSuchEntity(Entity.EntityType.TAG, tagIdentifier.name());
+ }
+ // The tag row is locked before any policy row, so every path through this
service takes its
+ // locks in the same tag-then-policy order.
+ return
TagMetaService.lockTags(Collections.singletonList(observed)).get(observed.getTagId());
}
private static Map<String, Long> resolvePolicyIds(
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/TagMetaService.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/TagMetaService.java
index 6726bc8de7..a704a74f11 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/service/TagMetaService.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/TagMetaService.java
@@ -21,9 +21,11 @@ package org.apache.gravitino.storage.relational.service;
import static
org.apache.gravitino.metrics.source.MetricsSource.GRAVITINO_RELATIONAL_STORE_METRIC_NAME;
import com.fasterxml.jackson.core.JsonProcessingException;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import java.io.IOException;
+import java.io.UncheckedIOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -49,9 +51,14 @@ import org.apache.gravitino.json.JsonUtils;
import org.apache.gravitino.meta.GenericEntity;
import org.apache.gravitino.meta.TagEntity;
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.PolicyMetadataObjectRelMapper;
import org.apache.gravitino.storage.relational.mapper.PolicyTagRelMapper;
+import org.apache.gravitino.storage.relational.mapper.SecurableObjectMapper;
import org.apache.gravitino.storage.relational.mapper.TagMetaMapper;
import
org.apache.gravitino.storage.relational.mapper.TagMetadataObjectRelMapper;
+import org.apache.gravitino.storage.relational.po.MetalakePO;
import org.apache.gravitino.storage.relational.po.TagMetadataObjectRelPO;
import org.apache.gravitino.storage.relational.po.TagPO;
import org.apache.gravitino.storage.relational.utils.ExceptionUtils;
@@ -100,20 +107,22 @@ public class TagMetaService {
String metalakeName = ns.level(0);
try {
- Long metalakeId =
MetalakeMetaService.getInstance().getMetalakeIdByName(metalakeName);
+ 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);
+ }
- TagPO.Builder builder = TagPO.builder().withMetalakeId(metalakeId);
+ TagPO.Builder builder =
TagPO.builder().withMetalakeId(metalakePO.getMetalakeId());
TagPO tagPO = POConverters.initializeTagPOWithVersion(tagEntity,
builder);
- SessionUtils.doWithCommit(
- TagMetaMapper.class,
- mapper -> {
- if (overwritten) {
- mapper.insertTagMetaOnDuplicateKeyUpdate(tagPO);
- } else {
- mapper.insertTagMeta(tagPO);
- }
- });
+ SessionUtils.doMultipleWithCommit(
+ () -> lockMetalakeForTagCreate(metalakePO),
+ () -> insertTagWithoutCommit(tagEntity, tagPO, overwritten));
} catch (RuntimeException e) {
ExceptionUtils.checkSQLException(e, Entity.EntityType.TAG,
tagEntity.toString());
throw e;
@@ -143,7 +152,7 @@ public class TagMetaService {
POConverters.updateTagPOWithVersion(tagPO,
updatedTagEntity), tagPO));
if (result == null || result == 0) {
- throw new IOException("Failed to update the entity: " + identifier);
+ throw tagWriteFailure(identifier, tagPO);
}
return updatedTagEntity;
@@ -156,27 +165,13 @@ public class TagMetaService {
@Monitored(metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
baseMetricName = "deleteTag")
public boolean deleteTag(NameIdentifier identifier) {
- String metalakeName = identifier.namespace().level(0);
- int[] tagDeletedCount = new int[] {0};
- int[] tagMetadataObjectRelDeletedCount = new int[] {0};
-
- SessionUtils.doMultipleWithCommit(
- () ->
- tagDeletedCount[0] =
- SessionUtils.getWithoutCommit(
- TagMetaMapper.class,
- mapper ->
- mapper.softDeleteTagMetaByMetalakeAndTagName(
- metalakeName, identifier.name())),
- () ->
- tagMetadataObjectRelDeletedCount[0] =
- SessionUtils.getWithoutCommit(
- TagMetadataObjectRelMapper.class,
- mapper ->
-
mapper.softDeleteTagMetadataObjectRelsByMetalakeAndTagName(
- metalakeName, identifier.name())));
-
- return tagDeletedCount[0] + tagMetadataObjectRelDeletedCount[0] > 0;
+ TagPO tagPO;
+ try {
+ tagPO = getTagPOByMetalakeAndName(identifier.namespace().level(0),
identifier.name());
+ } catch (NoSuchEntityException e) {
+ return false;
+ }
+ return deleteTag(identifier, tagPO);
}
@Monitored(
@@ -342,6 +337,14 @@ public class TagMetaService {
false /* failOnDuplicateValuelessAssignment */);
}
+ /**
+ * Runs one assignment change in a single transaction, so the tag rows stay
locked from the moment
+ * they are read until the relation rows are rewritten and read back. A
conflict rolls the whole
+ * change back instead of leaving a half-applied assignment set behind.
+ *
+ * <p>The mapper handed to the callback is unused: the call is only here to
open and close the
+ * transaction around work that talks to several mappers.
+ */
private List<TagEntity> associateTagValuesWithMetadataObject(
NameIdentifier objectIdent,
Entity.EntityType objectType,
@@ -349,6 +352,35 @@ public class TagMetaService {
TagValue[] tagsToRemove,
boolean failOnDuplicateValuelessAssignment)
throws NoSuchEntityException, EntityAlreadyExistsException, IOException {
+ try {
+ return SessionUtils.doWithCommitAndFetchResult(
+ TagMetaMapper.class,
+ ignored -> {
+ try {
+ return associateTagValuesWithMetadataObjectWithoutCommit(
+ objectIdent,
+ objectType,
+ tagsToAdd,
+ tagsToRemove,
+ failOnDuplicateValuelessAssignment);
+ } catch (IOException e) {
+ // The callback cannot throw a checked exception, so the
IOException raised while
+ // reading a tag's allowed values is carried across the boundary
and unwrapped below.
+ throw new UncheckedIOException(e);
+ }
+ });
+ } catch (UncheckedIOException e) {
+ throw e.getCause();
+ }
+ }
+
+ private List<TagEntity> associateTagValuesWithMetadataObjectWithoutCommit(
+ NameIdentifier objectIdent,
+ Entity.EntityType objectType,
+ TagValue[] tagsToAdd,
+ TagValue[] tagsToRemove,
+ boolean failOnDuplicateValuelessAssignment)
+ throws NoSuchEntityException, EntityAlreadyExistsException, IOException {
MetadataObject metadataObject =
NameIdentifierUtil.toMetadataObject(objectIdent, objectType);
String metalake = objectIdent.namespace().level(0);
@@ -366,6 +398,7 @@ public class TagMetaService {
tagNamesToUpdate.isEmpty()
? Collections.emptyList()
: getTagPOsByMetalakeAndNames(metalake, tagNamesToUpdate);
+ tagPOsToUpdate = lockTagsForAssignment(tagPOsToUpdate);
Map<String, TagPO> tagPOsByName = tagPOsByName(tagPOsToUpdate);
List<TagPO> currentTagPOs =
@@ -451,38 +484,25 @@ public class TagMetaService {
tagValueToAdd.value().orElse(null)));
}
- SessionUtils.doMultipleWithCommit(
- () -> {
- if (tagIdsToRemove.isEmpty()) {
- return;
- }
-
- SessionUtils.doWithoutCommit(
- TagMetadataObjectRelMapper.class,
- mapper ->
-
mapper.batchDeleteTagMetadataObjectRelsByTagIdsAndMetadataObject(
- metadataObjectId, metadataObject.type().toString(),
tagIdsToRemove));
- },
- () -> {
- if (tagRelsToRemove.isEmpty()) {
- return;
- }
-
- SessionUtils.doWithoutCommit(
- TagMetadataObjectRelMapper.class,
- mapper ->
-
mapper.batchDeleteTagMetadataObjectRelsByTagIdsAndValuesAndMetadataObject(
- metadataObjectId, metadataObject.type().toString(),
tagRelsToRemove));
- },
- () -> {
- if (tagRelsToAdd.isEmpty()) {
- return;
- }
-
- SessionUtils.doWithoutCommit(
- TagMetadataObjectRelMapper.class,
- mapper ->
mapper.batchInsertTagMetadataObjectRels(tagRelsToAdd));
- });
+ if (!tagIdsToRemove.isEmpty()) {
+ SessionUtils.doWithoutCommit(
+ TagMetadataObjectRelMapper.class,
+ mapper ->
+
mapper.batchDeleteTagMetadataObjectRelsByTagIdsAndMetadataObject(
+ metadataObjectId, metadataObject.type().toString(),
tagIdsToRemove));
+ }
+ if (!tagRelsToRemove.isEmpty()) {
+ SessionUtils.doWithoutCommit(
+ TagMetadataObjectRelMapper.class,
+ mapper ->
+
mapper.batchDeleteTagMetadataObjectRelsByTagIdsAndValuesAndMetadataObject(
+ metadataObjectId, metadataObject.type().toString(),
tagRelsToRemove));
+ }
+ if (!tagRelsToAdd.isEmpty()) {
+ SessionUtils.doWithoutCommit(
+ TagMetadataObjectRelMapper.class,
+ mapper -> mapper.batchInsertTagMetadataObjectRels(tagRelsToAdd));
+ }
List<TagPO> tagPOs =
SessionUtils.getWithoutCommit(
@@ -527,6 +547,61 @@ public class TagMetaService {
return tagDeletedCount[0] + tagMetadataObjectRelDeletedCount[0] +
policyTagRelDeletedCount[0];
}
+ /**
+ * Deletes the tag the caller observed. The delete is a compare-and-set on
the observed version,
+ * so a tag that changed since it was read is rejected instead of being
removed, and the dependent
+ * rows are cleaned up in the same transaction as the tag row itself. If
another transaction
+ * deletes or renames the observed tag first, this method returns {@code
false}, preserving the
+ * idempotent delete contract.
+ *
+ * <p>The observed row is a parameter so that a test can hand in a stale
one; production callers
+ * use {@link #deleteTag(NameIdentifier)}, which reads it first.
+ */
+ @VisibleForTesting
+ boolean deleteTag(NameIdentifier identifier, TagPO tagPO) {
+ long tagId = tagPO.getTagId();
+
+ // The version-checked delete of the tag row must stay first: it is what
decides whether this
+ // delete wins, and a losing delete throws there and rolls the transaction
back before any
+ // dependent row is touched. The cleanups that follow are blanket soft
deletes for this tag ID,
+ // so their row counts carry no information -- a tag with no assignments,
no policies, no owner
+ // and no securable object legitimately clears zero rows -- and only the
delete above is
+ // checked.
+ try {
+ SessionUtils.doMultipleWithCommit(
+ () -> deleteTagWithVersion(identifier, tagPO),
+ () ->
+ SessionUtils.doWithoutCommit(
+ TagMetadataObjectRelMapper.class,
+ mapper ->
mapper.softDeleteTagMetadataObjectRelsByTagId(tagId)),
+ () ->
+ SessionUtils.doWithoutCommit(
+ PolicyTagRelMapper.class, mapper ->
mapper.softDeleteByTagId(tagId)),
+ () ->
+ SessionUtils.doWithoutCommit(
+ PolicyMetadataObjectRelMapper.class,
+ mapper ->
+
mapper.softDeletePolicyMetadataObjectRelsByMetadataObject(
+ tagId, MetadataObject.Type.TAG.name())),
+ () ->
+ SessionUtils.doWithoutCommit(
+ OwnerMetaMapper.class,
+ mapper ->
+ mapper.softDeleteOwnerRelByMetadataObjectIdAndType(
+ tagId, MetadataObject.Type.TAG.name())),
+ () ->
+ SessionUtils.doWithoutCommit(
+ SecurableObjectMapper.class,
+ mapper ->
+ mapper.softDeleteObjectRelsByMetadataObject(
+ tagId, MetadataObject.Type.TAG.name())));
+ } catch (NoSuchEntityException e) {
+ return false;
+ }
+
+ return true;
+ }
+
private static List<TagEntity> tagPOsToTagEntities(List<TagPO> tagPOs,
Namespace namespace) {
Map<Long, List<TagPO>> tagPOsByTagId = new LinkedHashMap<>();
for (TagPO tagPO : tagPOs) {
@@ -625,6 +700,171 @@ public class TagMetaService {
Arrays.toString(allowedValues));
}
+ /**
+ * Holds the parent metalake row for the rest of the transaction, so a tag
cannot be created under
+ * a metalake that is going away.
+ *
+ * <p>The lock is shared, not exclusive: many tags can be created under the
same metalake at the
+ * same time, while dropping the metalake takes an exclusive lock on this
row, so a drop and a
+ * create cannot overlap.
+ *
+ * <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. The metalake version is deliberately not compared, matching
{@code CatalogMetaService}:
+ * holding the row is what makes the create safe, and an unrelated metalake
edit that commits in
+ * between would otherwise reject the create for no reason.
+ */
+ private 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()));
+ }
+
+ /**
+ * Writes a new tag or replaces the active tag selected by name or stable ID.
+ *
+ * <p>The overwrite path uses locking reads instead of a database-specific
upsert. This keeps the
+ * stored tag ID and OCC sequence stable across all databases. A
name-targeted overwrite that
+ * races with a rename creates a new row under the now-free name instead of
reverting the renamed
+ * row.
+ */
+ private void insertTagWithoutCommit(
+ TagEntity tagEntity, TagPO initializedTagPO, boolean overwritten) {
+ if (!overwritten) {
+ insertNewTagWithoutCommit(initializedTagPO);
+ return;
+ }
+
+ TagPO existingTagPO = findAndLockTagForOverwrite(initializedTagPO);
+ if (existingTagPO == null) {
+ insertNewTagWithoutCommit(initializedTagPO);
+ return;
+ }
+
+ TagPO replacementTagPO =
POConverters.updateTagPOWithVersion(existingTagPO, tagEntity);
+ NameIdentifier observedIdentifier =
+ NameIdentifier.of(tagEntity.namespace(), existingTagPO.getTagName());
+ updateTagRootWithVersion(observedIdentifier, existingTagPO,
replacementTagPO);
+ }
+
+ private void insertNewTagWithoutCommit(TagPO tagPO) {
+ SessionUtils.doWithoutCommit(TagMetaMapper.class, mapper ->
mapper.insertTagMeta(tagPO));
+ }
+
+ private TagPO findAndLockTagForOverwrite(TagPO initializedTagPO) {
+ TagPO sameNameTagPO =
+ SessionUtils.getWithoutCommit(
+ TagMetaMapper.class,
+ mapper ->
+ mapper.selectTagMetaByMetalakeIdAndNameForUpdate(
+ initializedTagPO.getMetalakeId(),
initializedTagPO.getTagName()));
+ if (sameNameTagPO != null) {
+ return sameNameTagPO;
+ }
+
+ TagPO sameIdTagPO =
+ SessionUtils.getWithoutCommit(
+ TagMetaMapper.class,
+ mapper ->
mapper.selectTagByTagIdForUpdate(initializedTagPO.getTagId()));
+ if (sameIdTagPO == null
+ || !Objects.equals(sameIdTagPO.getMetalakeId(),
initializedTagPO.getMetalakeId())) {
+ return null;
+ }
+ return sameIdTagPO;
+ }
+
+ private void updateTagRootWithVersion(NameIdentifier identifier, TagPO
oldTagPO, TagPO newTagPO) {
+ Integer updated =
+ SessionUtils.getWithoutCommit(
+ TagMetaMapper.class, mapper -> mapper.updateTagMeta(newTagPO,
oldTagPO));
+ if (updated == null || updated == 0) {
+ throw tagWriteFailure(identifier, oldTagPO);
+ }
+ }
+
+ 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()));
+ }
+
+ /**
+ * Locks every tag taking part in an assignment change and returns the rows
as they are now, so
+ * the assignment cannot be written against a tag that is being renamed or
dropped.
+ *
+ * <p>See {@link #lockTags} for how the rows are locked and what counts as a
missing tag.
+ */
+ private List<TagPO> lockTagsForAssignment(List<TagPO> observedTagPOs) {
+ return new ArrayList<>(lockTags(observedTagPOs).values());
+ }
+
+ /**
+ * Locks the given tag rows and returns them as they are now, keyed by tag
ID.
+ *
+ * <p>The rows are locked by one statement that orders them by tag ID, so
callers that touch
+ * overlapping tags take the row locks in the same order and queue up
instead of deadlocking, and
+ * a change touching many tags still costs a single round trip.
+ *
+ * <p>A row that is gone, or whose name or metalake no longer matches what
the caller resolved by
+ * name, is reported as missing: the caller asked for a tag name, and that
name no longer points
+ * at this row.
+ */
+ static Map<Long, TagPO> lockTags(List<TagPO> observedTagPOs) {
+ Map<Long, TagPO> observedById = new LinkedHashMap<>();
+ observedTagPOs.forEach(tagPO -> observedById.put(tagPO.getTagId(), tagPO));
+ if (observedById.isEmpty()) {
+ return new LinkedHashMap<>();
+ }
+
+ List<TagPO> lockedRows =
+ SessionUtils.getWithoutCommit(
+ TagMetaMapper.class,
+ mapper -> mapper.listTagPOsByTagIdsForUpdate(new
ArrayList<>(observedById.keySet())));
+ Map<Long, TagPO> lockedById = new LinkedHashMap<>();
+ lockedRows.forEach(tagPO -> lockedById.put(tagPO.getTagId(), tagPO));
+
+ Map<Long, TagPO> lockedTagPOs = new LinkedHashMap<>();
+ for (TagPO observedTagPO : observedById.values()) {
+ TagPO lockedTagPO =
+ OccWriteSupport.lockParentForChildWrite(
+ observedTagPO.getTagName(),
+ Entity.EntityType.TAG,
+ () -> lockedById.get(observedTagPO.getTagId()),
+ null,
+ current ->
+ Objects.equals(current.getTagName(),
observedTagPO.getTagName())
+ && Objects.equals(current.getMetalakeId(),
observedTagPO.getMetalakeId()));
+ lockedTagPOs.put(lockedTagPO.getTagId(), lockedTagPO);
+ }
+ return lockedTagPOs;
+ }
+
private TagPO getTagPOByMetalakeAndName(String metalakeName, String tagName)
{
TagPO tagPO =
SessionUtils.getWithoutCommit(
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 51f21421cf..07f98776c8 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
@@ -1423,10 +1423,7 @@ 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;
+ Long nextVersion = oldTagPO.getCurrentVersion() + 1;
try {
return TagPO.builder()
.withTagId(oldTagPO.getTagId())
diff --git a/core/src/main/java/org/apache/gravitino/tag/TagManager.java
b/core/src/main/java/org/apache/gravitino/tag/TagManager.java
index 27ccd5a518..69bc1c4d62 100644
--- a/core/src/main/java/org/apache/gravitino/tag/TagManager.java
+++ b/core/src/main/java/org/apache/gravitino/tag/TagManager.java
@@ -46,6 +46,7 @@ import org.apache.gravitino.exceptions.NoSuchEntityException;
import org.apache.gravitino.exceptions.NoSuchMetadataObjectException;
import org.apache.gravitino.exceptions.NoSuchTagException;
import org.apache.gravitino.exceptions.NotFoundException;
+import org.apache.gravitino.exceptions.OptimisticLockException;
import org.apache.gravitino.exceptions.PolicyAlreadyAssociatedException;
import org.apache.gravitino.exceptions.TagAlreadyAssociatedException;
import org.apache.gravitino.exceptions.TagAlreadyExistsException;
@@ -192,6 +193,16 @@ public class TagManager implements TagDispatcher {
.orElse(name);
throw new TagAlreadyExistsException(
e, "Tag with name %s under metalake %s already exists",
newName, metalake);
+ } catch (OptimisticLockException ole) {
+ // The store now rejects a stale alter with this exception instead
of an IOException,
+ // and the REST layer maps it to a conflict. Log it here so the
operator-facing record
+ // still names the tag and the metalake.
+ LOG.warn(
+ "Failed to alter tag {} under metalake {} because it changed
concurrently",
+ name,
+ metalake,
+ ole);
+ throw ole;
} catch (IOException ioe) {
LOG.error("Failed to alter tag {} under metalake {}", name,
metalake, ioe);
throw new RuntimeException(ioe);
@@ -208,6 +219,13 @@ public class TagManager implements TagDispatcher {
try {
return entityStore.delete(
NameIdentifierUtil.ofTag(metalake, name),
Entity.EntityType.TAG);
+ } catch (OptimisticLockException ole) {
+ LOG.warn(
+ "Failed to delete tag {} under metalake {} because it changed
concurrently",
+ name,
+ metalake,
+ ole);
+ throw ole;
} catch (IOException ioe) {
LOG.error("Failed to delete tag {} under metalake {}", name,
metalake, ioe);
throw new RuntimeException(ioe);
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestTagMetaService.java
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestTagMetaService.java
index f263a68e12..b971030a85 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestTagMetaService.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestTagMetaService.java
@@ -23,6 +23,7 @@ import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Lists;
import java.io.IOException;
import java.sql.Connection;
@@ -34,30 +35,52 @@ import java.util.Arrays;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import org.apache.gravitino.Entity;
import org.apache.gravitino.EntityAlreadyExistsException;
+import org.apache.gravitino.MetadataObject;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.Namespace;
import org.apache.gravitino.RelationEdgeTarget;
import org.apache.gravitino.RelationQuery;
import org.apache.gravitino.RelationUpdate;
import org.apache.gravitino.SupportsRelationOperations;
+import org.apache.gravitino.authorization.AuthorizationUtils;
+import org.apache.gravitino.authorization.Privileges;
+import org.apache.gravitino.authorization.SecurableObjects;
import org.apache.gravitino.exceptions.NoSuchEntityException;
+import org.apache.gravitino.exceptions.OptimisticLockException;
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.GenericEntity;
import org.apache.gravitino.meta.ModelEntity;
+import org.apache.gravitino.meta.PolicyEntity;
+import org.apache.gravitino.meta.RoleEntity;
import org.apache.gravitino.meta.SchemaEntity;
import org.apache.gravitino.meta.TableEntity;
import org.apache.gravitino.meta.TagEntity;
import org.apache.gravitino.meta.TopicEntity;
+import org.apache.gravitino.meta.UserEntity;
+import org.apache.gravitino.policy.PolicyContents;
import org.apache.gravitino.rel.types.Types;
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.TagMetaMapper;
+import org.apache.gravitino.storage.relational.po.MetalakePO;
+import org.apache.gravitino.storage.relational.po.TagPO;
import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
+import org.apache.gravitino.storage.relational.utils.POConverters;
+import org.apache.gravitino.storage.relational.utils.SessionUtils;
import org.apache.gravitino.tag.TagValue;
+import org.apache.gravitino.tag.TagValueConstraint;
import org.apache.gravitino.utils.NameIdentifierUtil;
import org.apache.gravitino.utils.NamespaceUtil;
import org.apache.ibatis.session.SqlSession;
@@ -321,6 +344,526 @@ public class TestTagMetaService extends TestJDBCBackend {
Assertions.assertEquals(tagEntity2, loadedTagEntity1);
}
+ @TestTemplate
+ public void testTagAlterDeleteAndOverwriteUseMonotonicVersion() throws
IOException {
+ createAndInsertMakeLake(METALAKE_NAME);
+ TagMetaService tagMetaService = TagMetaService.getInstance();
+ TagEntity tag =
+ TagEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName("tag_occ")
+ .withNamespace(NamespaceUtil.ofTag(METALAKE_NAME))
+ .withComment("initial")
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ tagMetaService.insertTag(tag, false);
+
+ TagPO initialPO =
+ SessionUtils.getWithoutCommit(
+ TagMetaMapper.class, mapper -> mapper.selectTagByTagId(tag.id()));
+ TagEntity replacement =
+ TagEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName(tag.name())
+ .withNamespace(tag.namespace())
+ .withComment("overwritten")
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ tagMetaService.insertTag(replacement, true);
+ TagPO overwrittenPO =
+ SessionUtils.getWithoutCommit(
+ TagMetaMapper.class, mapper -> mapper.selectTagByTagId(tag.id()));
+ Assertions.assertEquals(tag.id(), overwrittenPO.getTagId().longValue());
+ Assertions.assertEquals("overwritten", overwrittenPO.getComment());
+ Assertions.assertEquals(
+ initialPO.getCurrentVersion() + 1,
overwrittenPO.getCurrentVersion().longValue());
+ Assertions.assertEquals(overwrittenPO.getCurrentVersion(),
overwrittenPO.getLastVersion());
+
+ TagEntity updatedTag = copyTagWithComment(tag, "updated");
+ TagPO nextPO = POConverters.updateTagPOWithVersion(overwrittenPO,
updatedTag);
+ Assertions.assertEquals(
+ Integer.valueOf(1),
+ SessionUtils.doWithCommitAndFetchResult(
+ TagMetaMapper.class, mapper -> mapper.updateTagMeta(nextPO,
overwrittenPO)));
+ Assertions.assertEquals(
+ Integer.valueOf(0),
+ SessionUtils.doWithCommitAndFetchResult(
+ TagMetaMapper.class, mapper -> mapper.updateTagMeta(nextPO,
overwrittenPO)));
+ Assertions.assertEquals(
+ Integer.valueOf(0),
+ SessionUtils.doWithCommitAndFetchResult(
+ TagMetaMapper.class,
+ mapper ->
+ mapper.softDeleteTagMetaByIdAndVersion(
+ tag.id(), overwrittenPO.getCurrentVersion())));
+ Assertions.assertEquals(
+ Integer.valueOf(1),
+ SessionUtils.doWithCommitAndFetchResult(
+ TagMetaMapper.class,
+ mapper ->
+ mapper.softDeleteTagMetaByIdAndVersion(tag.id(),
nextPO.getCurrentVersion())));
+ }
+
+ @TestTemplate
+ public void testTagOverwriteByNameDoesNotRevertConcurrentRename() throws
Exception {
+ createAndInsertMakeLake(METALAKE_NAME);
+ TagMetaService tagMetaService = TagMetaService.getInstance();
+ TagEntity original =
+ TagEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName("tag_overwrite_rename_race")
+ .withNamespace(NamespaceUtil.ofTag(METALAKE_NAME))
+ .withComment("initial")
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ tagMetaService.insertTag(original, false);
+ TagPO observedPO = getTagPO(original.nameIdentifier());
+ TagEntity renamed = copyTag(original, "tag_overwrite_rename_winner",
"rename winner");
+ TagPO renamedPO = POConverters.updateTagPOWithVersion(observedPO, renamed);
+ TagEntity replacement =
+ TagEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName(original.name())
+ .withNamespace(original.namespace())
+ .withComment("replacement")
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+
+ CountDownLatch renameWritten = new CountDownLatch(1);
+ CountDownLatch allowRenameCommit = new CountDownLatch(1);
+ CountDownLatch overwriteStarted = new CountDownLatch(1);
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ Future<Throwable> renameResult =
+ executor.submit(
+ () -> {
+ try {
+ SessionUtils.doMultipleWithCommit(
+ () ->
+ Assertions.assertEquals(
+ Integer.valueOf(1),
+ SessionUtils.getWithoutCommit(
+ TagMetaMapper.class,
+ mapper -> mapper.updateTagMeta(renamedPO,
observedPO))),
+ () -> {
+ renameWritten.countDown();
+ await(allowRenameCommit);
+ });
+ return null;
+ } catch (Throwable throwable) {
+ return throwable;
+ }
+ });
+
+ try {
+ assertTrue(renameWritten.await(30, TimeUnit.SECONDS));
+ Future<Throwable> overwriteResult =
+ executor.submit(
+ () -> {
+ overwriteStarted.countDown();
+ try {
+ tagMetaService.insertTag(replacement, true);
+ return null;
+ } catch (Throwable throwable) {
+ return throwable;
+ }
+ });
+ assertTrue(overwriteStarted.await(30, TimeUnit.SECONDS));
+ Assertions.assertThrows(
+ TimeoutException.class, () -> overwriteResult.get(500,
TimeUnit.MILLISECONDS));
+
+ allowRenameCommit.countDown();
+ Assertions.assertNull(renameResult.get(30, TimeUnit.SECONDS));
+ Assertions.assertNull(overwriteResult.get(30, TimeUnit.SECONDS));
+ } finally {
+ allowRenameCommit.countDown();
+ executor.shutdownNow();
+ }
+
+ Assertions.assertEquals(
+ original.id(),
tagMetaService.getTagByIdentifier(renamed.nameIdentifier()).id());
+ Assertions.assertEquals(
+ replacement.id(),
tagMetaService.getTagByIdentifier(replacement.nameIdentifier()).id());
+ }
+
+ @TestTemplate
+ public void testTagCreateWaitsForConcurrentParentDelete() throws Exception {
+ BaseMetalake metalake = createAndInsertMakeLake(METALAKE_NAME);
+ MetalakePO observedMetalakePO =
+ SessionUtils.getWithoutCommit(
+ MetalakeMetaMapper.class, mapper ->
mapper.selectMetalakeMetaByName(metalake.name()));
+ TagEntity tag =
+ TagEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName("tag_parent_delete_race")
+ .withNamespace(NamespaceUtil.ofTag(METALAKE_NAME))
+ .withComment("initial")
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+
+ CountDownLatch deleteWritten = new CountDownLatch(1);
+ CountDownLatch allowDeleteCommit = new CountDownLatch(1);
+ CountDownLatch createStarted = new CountDownLatch(1);
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ Future<Throwable> deleteResult =
+ executor.submit(
+ () -> {
+ try {
+ SessionUtils.doMultipleWithCommit(
+ () ->
+ Assertions.assertEquals(
+ Integer.valueOf(1),
+ SessionUtils.getWithoutCommit(
+ MetalakeMetaMapper.class,
+ mapper ->
+ mapper.softDeleteMetalakeMetaByMetalakeId(
+ observedMetalakePO.getMetalakeId(),
+
observedMetalakePO.getCurrentVersion()))),
+ () -> {
+ deleteWritten.countDown();
+ await(allowDeleteCommit);
+ });
+ return null;
+ } catch (Throwable throwable) {
+ return throwable;
+ }
+ });
+
+ try {
+ assertTrue(deleteWritten.await(30, TimeUnit.SECONDS));
+ Future<Throwable> createResult =
+ executor.submit(
+ () -> {
+ createStarted.countDown();
+ try {
+ TagMetaService.getInstance().insertTag(tag, false);
+ return null;
+ } catch (Throwable throwable) {
+ return throwable;
+ }
+ });
+ assertTrue(createStarted.await(30, TimeUnit.SECONDS));
+ Assertions.assertThrows(
+ TimeoutException.class, () -> createResult.get(500,
TimeUnit.MILLISECONDS));
+
+ allowDeleteCommit.countDown();
+ Assertions.assertNull(deleteResult.get(30, TimeUnit.SECONDS));
+ Throwable createFailure = createResult.get(30, TimeUnit.SECONDS);
+ Assertions.assertTrue(
+ createFailure instanceof NoSuchEntityException,
String.valueOf(createFailure));
+ } finally {
+ allowDeleteCommit.countDown();
+ executor.shutdownNow();
+ }
+
+ assertFalse(backend.exists(tag.nameIdentifier(), Entity.EntityType.TAG));
+ Assertions.assertNull(
+ SessionUtils.getWithoutCommit(
+ TagMetaMapper.class, mapper -> mapper.selectTagByTagId(tag.id())));
+ }
+
+ @TestTemplate
+ public void testTagAssignmentWaitsForConcurrentRenameAndRollsBack() throws
Exception {
+ createAndInsertMakeLake(METALAKE_NAME);
+ CatalogEntity catalog = createAndInsertCatalog(METALAKE_NAME,
"catalog_tag_rename_race");
+ TagMetaService tagMetaService = TagMetaService.getInstance();
+ TagEntity tag =
+ TagEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName("tag_assignment_rename_race")
+ .withNamespace(NamespaceUtil.ofTag(METALAKE_NAME))
+ .withComment("initial")
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ tagMetaService.insertTag(tag, false);
+ TagPO observedPO = getTagPO(tag.nameIdentifier());
+ TagEntity renamed = copyTag(tag, "tag_assignment_rename_winner",
"renamed");
+ TagPO renamedPO = POConverters.updateTagPOWithVersion(observedPO, renamed);
+
+ CountDownLatch renameWritten = new CountDownLatch(1);
+ CountDownLatch allowRenameCommit = new CountDownLatch(1);
+ CountDownLatch assignmentStarted = new CountDownLatch(1);
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ Future<Throwable> renameResult =
+ executor.submit(
+ () -> {
+ try {
+ SessionUtils.doMultipleWithCommit(
+ () ->
+ Assertions.assertEquals(
+ Integer.valueOf(1),
+ SessionUtils.getWithoutCommit(
+ TagMetaMapper.class,
+ mapper -> mapper.updateTagMeta(renamedPO,
observedPO))),
+ () -> {
+ renameWritten.countDown();
+ await(allowRenameCommit);
+ });
+ return null;
+ } catch (Throwable throwable) {
+ return throwable;
+ }
+ });
+
+ try {
+ assertTrue(renameWritten.await(30, TimeUnit.SECONDS));
+ Future<Throwable> assignmentResult =
+ executor.submit(
+ () -> {
+ assignmentStarted.countDown();
+ try {
+ tagMetaService.associateTagsWithMetadataObject(
+ catalog.nameIdentifier(),
+ catalog.type(),
+ new NameIdentifier[] {tag.nameIdentifier()},
+ new NameIdentifier[0]);
+ return null;
+ } catch (Throwable throwable) {
+ return throwable;
+ }
+ });
+ assertTrue(assignmentStarted.await(30, TimeUnit.SECONDS));
+ Assertions.assertThrows(
+ TimeoutException.class, () -> assignmentResult.get(500,
TimeUnit.MILLISECONDS));
+
+ allowRenameCommit.countDown();
+ Assertions.assertNull(renameResult.get(30, TimeUnit.SECONDS));
+ Throwable assignmentFailure = assignmentResult.get(30, TimeUnit.SECONDS);
+ Assertions.assertTrue(
+ assignmentFailure instanceof NoSuchEntityException,
String.valueOf(assignmentFailure));
+ } finally {
+ allowRenameCommit.countDown();
+ executor.shutdownNow();
+ }
+
+ assertEquals(0, countActiveTagRel(tag.id()));
+ Assertions.assertEquals(
+ tag.id(),
tagMetaService.getTagByIdentifier(renamed.nameIdentifier()).id());
+ }
+
+ @TestTemplate
+ public void testTagAlterReportsOptimisticLockConflict() throws IOException {
+ createAndInsertMakeLake(METALAKE_NAME);
+ TagMetaService tagMetaService = TagMetaService.getInstance();
+ TagEntity tag =
+ TagEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName("tag_alter_conflict")
+ .withNamespace(NamespaceUtil.ofTag(METALAKE_NAME))
+ .withComment("initial")
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ tagMetaService.insertTag(tag, false);
+
+ Assertions.assertThrows(
+ OptimisticLockException.class,
+ () ->
+ tagMetaService.updateTag(
+ tag.nameIdentifier(),
+ entity -> {
+ TagEntity current = (TagEntity) entity;
+ TagPO currentPO =
+ SessionUtils.getWithoutCommit(
+ TagMetaMapper.class, mapper ->
mapper.selectTagByTagId(current.id()));
+ TagPO competingPO =
+ POConverters.updateTagPOWithVersion(
+ currentPO, copyTagWithComment(current, "competing"));
+ SessionUtils.doWithCommitAndFetchResult(
+ TagMetaMapper.class, mapper ->
mapper.updateTagMeta(competingPO, currentPO));
+ return copyTagWithComment(current, "requested");
+ }));
+ }
+
+ @TestTemplate
+ public void testStaleTagDeleteRollsBackRelationshipCleanup() throws
IOException {
+ createAndInsertMakeLake(METALAKE_NAME);
+ CatalogEntity catalog = createAndInsertCatalog(METALAKE_NAME,
"catalog_tag_delete_occ");
+ TagMetaService tagMetaService = TagMetaService.getInstance();
+ TagEntity tag =
+ TagEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName("tag_delete_occ")
+ .withNamespace(NamespaceUtil.ofTag(METALAKE_NAME))
+ .withComment("initial")
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ tagMetaService.insertTag(tag, false);
+ tagMetaService.associateTagsWithMetadataObject(
+ catalog.nameIdentifier(),
+ catalog.type(),
+ new NameIdentifier[] {tag.nameIdentifier()},
+ new NameIdentifier[0]);
+ TagPO stalePO =
+ SessionUtils.getWithoutCommit(
+ TagMetaMapper.class, mapper -> mapper.selectTagByTagId(tag.id()));
+ tagMetaService.updateTag(
+ tag.nameIdentifier(), entity -> copyTagWithComment((TagEntity) entity,
"updated"));
+
+ Assertions.assertThrows(
+ OptimisticLockException.class,
+ () -> tagMetaService.deleteTag(tag.nameIdentifier(), stalePO));
+ Assertions.assertEquals(1, countActiveTagRel(tag.id()));
+ Assertions.assertTrue(backend.exists(tag.nameIdentifier(),
Entity.EntityType.TAG));
+
+ Assertions.assertTrue(tagMetaService.deleteTag(tag.nameIdentifier()));
+ Assertions.assertEquals(0, countActiveTagRel(tag.id()));
+ }
+
+ @TestTemplate
+ public void testTagCreateIsFencedByParentMetalake() {
+ TagMetaService tagMetaService = TagMetaService.getInstance();
+ TagEntity tag =
+ TagEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName("tag_without_metalake")
+ .withNamespace(NamespaceUtil.ofTag("metalake_that_does_not_exist"))
+ .withComment("initial")
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+
+ Assertions.assertThrows(
+ NoSuchEntityException.class, () -> tagMetaService.insertTag(tag,
false));
+ Assertions.assertThrows(NoSuchEntityException.class, () ->
tagMetaService.insertTag(tag, true));
+ }
+
+ @TestTemplate
+ public void testTagOverwriteAndAlterKeepAllowedValues() throws IOException {
+ createAndInsertMakeLake(METALAKE_NAME);
+ TagMetaService tagMetaService = TagMetaService.getInstance();
+ TagEntity tag =
+ TagEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName("tag_allowed_values_occ")
+ .withNamespace(NamespaceUtil.ofTag(METALAKE_NAME))
+ .withComment("initial")
+ .withAllowedValues(new String[] {"dev", "prod"})
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ tagMetaService.insertTag(tag, false);
+
+ tagMetaService.insertTag(copyTagWithComment(tag, "overwritten"), true);
+ TagEntity overwritten =
tagMetaService.getTagByIdentifier(tag.nameIdentifier());
+ Assertions.assertArrayEquals(
+ new String[] {"dev", "prod"},
overwritten.valueConstraint().allowedValues());
+
+ tagMetaService.updateTag(
+ tag.nameIdentifier(), entity -> copyTagWithComment((TagEntity) entity,
"updated"));
+ TagEntity updated =
tagMetaService.getTagByIdentifier(tag.nameIdentifier());
+ Assertions.assertEquals("updated", updated.comment());
+ Assertions.assertArrayEquals(
+ new String[] {"dev", "prod"},
updated.valueConstraint().allowedValues());
+ }
+
+ @TestTemplate
+ public void testDeleteTagCleansEveryDependentRelation() throws IOException {
+ createAndInsertMakeLake(METALAKE_NAME);
+ CatalogEntity catalog = createAndInsertCatalog(METALAKE_NAME,
"catalog_tag_cascade");
+ TagMetaService tagMetaService = TagMetaService.getInstance();
+ TagEntity tag =
+ TagEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName("tag_cascade_occ")
+ .withNamespace(NamespaceUtil.ofTag(METALAKE_NAME))
+ .withComment("initial")
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ tagMetaService.insertTag(tag, false);
+ tagMetaService.associateTagsWithMetadataObject(
+ catalog.nameIdentifier(),
+ catalog.type(),
+ new NameIdentifier[] {tag.nameIdentifier()},
+ new NameIdentifier[0]);
+
+ PolicyEntity policy =
+ createAndInsertPolicyEntity(
+ "policy_tag_cascade",
+ "policy comment",
+ PolicyContents.custom(
+ ImmutableMap.of("k", "v"),
ImmutableSet.of(MetadataObject.Type.TAG), null),
+ METALAKE_NAME);
+ backend.updateEntityRelations(
+ RelationUpdate.of(
+ SupportsRelationOperations.Type.POLICY_TAG_REL,
+ tag.nameIdentifier(),
+ Entity.EntityType.TAG,
+ new RelationEdgeTarget[] {
+ RelationEdgeTarget.of(
+ policy.nameIdentifier(),
+ Entity.EntityType.POLICY,
+ "{\"type\":\"TAG_VALUE\",\"value\":\"finance\"}")
+ },
+ new RelationEdgeTarget[0]));
+ PolicyMetaService.getInstance()
+ .associatePoliciesWithMetadataObject(
+ tag.nameIdentifier(),
+ Entity.EntityType.TAG,
+ new NameIdentifier[] {policy.nameIdentifier()},
+ new NameIdentifier[0]);
+
+ UserEntity user =
+ createUserEntity(
+ RandomIdGenerator.INSTANCE.nextId(),
+ AuthorizationUtils.ofUserNamespace(METALAKE_NAME),
+ "user_tag_cascade",
+ AUDIT_INFO);
+ backend.insert(user, false);
+ OwnerMetaService.getInstance()
+ .setOwner(tag.nameIdentifier(), Entity.EntityType.TAG,
user.nameIdentifier(), user.type());
+
+ RoleEntity role =
+ createRoleEntity(
+ RandomIdGenerator.INSTANCE.nextId(),
+ AuthorizationUtils.ofRoleNamespace(METALAKE_NAME),
+ "role_tag_cascade",
+ AUDIT_INFO,
+ Lists.newArrayList(
+ SecurableObjects.ofTag(
+ tag.name(),
Lists.newArrayList(Privileges.ApplyTag.allow()))),
+ null);
+ backend.insert(role, false);
+
+ String tagAsMetadataObject =
+ String.format("metadata_object_id = %d AND metadata_object_type =
'TAG'", tag.id());
+ String tagAsSecurableObject =
+ String.format("metadata_object_id = %d AND type = 'TAG'", tag.id());
+ assertEquals(1, countActiveTagRel(tag.id()));
+ assertEquals(1, countActiveRows("policy_tag_relation_meta", "tag_id = " +
tag.id()));
+ assertEquals(1, countActiveRows("policy_relation_meta",
tagAsMetadataObject));
+ assertEquals(1, countActiveRows("owner_meta", tagAsMetadataObject));
+ assertEquals(1, countActiveRows("role_meta_securable_object",
tagAsSecurableObject));
+
+ assertTrue(tagMetaService.deleteTag(tag.nameIdentifier()));
+
+ assertEquals(0, countActiveTagRel(tag.id()));
+ assertEquals(0, countActiveRows("policy_tag_relation_meta", "tag_id = " +
tag.id()));
+ assertEquals(0, countActiveRows("policy_relation_meta",
tagAsMetadataObject));
+ assertEquals(0, countActiveRows("owner_meta", tagAsMetadataObject));
+ assertEquals(0, countActiveRows("role_meta_securable_object",
tagAsSecurableObject));
+ }
+
+ @TestTemplate
+ public void testDeleteOfAlreadyDeletedTagReportsMissingTagNotConflict()
throws IOException {
+ createAndInsertMakeLake(METALAKE_NAME);
+ TagMetaService tagMetaService = TagMetaService.getInstance();
+ TagEntity tag =
+ TagEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName("tag_gone_occ")
+ .withNamespace(NamespaceUtil.ofTag(METALAKE_NAME))
+ .withComment("initial")
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ tagMetaService.insertTag(tag, false);
+ TagPO observedPO =
+ SessionUtils.getWithoutCommit(
+ TagMetaMapper.class, mapper -> mapper.selectTagByTagId(tag.id()));
+ Assertions.assertTrue(tagMetaService.deleteTag(tag.nameIdentifier()));
+
+ // The row the caller observed is gone rather than merely moved on, so the
idempotent delete
+ // contract reports false instead of turning a concurrent delete into an
exception.
+ Assertions.assertFalse(tagMetaService.deleteTag(tag.nameIdentifier(),
observedPO));
+ Assertions.assertFalse(tagMetaService.deleteTag(tag.nameIdentifier()));
+ }
+
@TestTemplate
public void testDeleteTag() throws IOException {
createAndInsertMakeLake(METALAKE_NAME);
@@ -1284,6 +1827,42 @@ public class TestTagMetaService extends TestJDBCBackend {
() -> tagMetaService.getTagIdByTagName(metalakeId, "missing_tag"));
}
+ private TagEntity copyTagWithComment(TagEntity tag, String comment) {
+ return copyTag(tag, tag.name(), comment);
+ }
+
+ private TagEntity copyTag(TagEntity tag, String name, String comment) {
+ TagEntity.Builder builder =
+ TagEntity.builder()
+ .withId(tag.id())
+ .withName(name)
+ .withNamespace(tag.namespace())
+ .withComment(comment)
+ .withProperties(tag.properties())
+ .withAuditInfo(tag.auditInfo());
+ if (tag.valueConstraint().type() != TagValueConstraint.Type.ANY_VALUE) {
+ builder.withAllowedValues(tag.valueConstraint().allowedValues());
+ }
+ return builder.build();
+ }
+
+ private TagPO getTagPO(NameIdentifier identifier) {
+ return SessionUtils.getWithoutCommit(
+ TagMetaMapper.class,
+ mapper ->
+ mapper.selectTagMetaByMetalakeAndName(
+ identifier.namespace().level(0), identifier.name()));
+ }
+
+ private void await(CountDownLatch latch) {
+ try {
+ assertTrue(latch.await(30, TimeUnit.SECONDS));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ }
+
private boolean containsValuelessTagAssignment(
List<TagEntity> tagEntities, TagEntity expectedTagEntity) {
return tagEntities.stream()
@@ -1307,6 +1886,24 @@ public class TestTagMetaService extends TestJDBCBackend {
return genericEntities.stream().anyMatch(e -> e.name().equals(name) &&
e.type() == entityType);
}
+ private int countActiveRows(String table, String whereClause) {
+ try (SqlSession sqlSession =
+
SqlSessionFactoryHelper.getInstance().getSqlSessionFactory().openSession(true);
+ Connection connection = sqlSession.getConnection();
+ Statement statement = connection.createStatement();
+ ResultSet rs =
+ statement.executeQuery(
+ String.format(
+ "SELECT count(*) FROM %s WHERE %s AND deleted_at = 0",
table, whereClause))) {
+ if (rs.next()) {
+ return rs.getInt(1);
+ }
+ throw new RuntimeException("Doesn't contain data");
+ } catch (SQLException se) {
+ throw new RuntimeException("SQL execution failed", se);
+ }
+ }
+
private Integer countAllTagRel(Long tagId) {
try (SqlSession sqlSession =
SqlSessionFactoryHelper.getInstance().getSqlSessionFactory().openSession(true);
