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 f47d6d1155 [#12776] feat(core): Add OCC for policy metadata (#12782)
f47d6d1155 is described below
commit f47d6d1155db57a5b2bb70075da305d789d4b063
Author: Qi Yu <[email protected]>
AuthorDate: Tue Sep 8 16:04:23 2026 +0800
[#12776] feat(core): Add OCC for policy metadata (#12782)
### What changes were proposed in this pull request?
- Apply stable-ID and expected-version CAS to policy alter, overwrite,
and delete operations.
- Create a complete immutable policy snapshot for every new version.
- Preserve monotonic policy version history during overwrite.
- Lock parent metalakes and policy roots during relationship changes.
- Atomically clean policy versions and dependent relationships during
deletion.
- Reuse the common OCC support introduced by #12639.
### Why are the changes needed?
Concurrent policy writes could otherwise cause lost updates, incomplete
snapshots, orphan versions, or partial relationship cleanup.
Fix: #12776
### Does this PR introduce _any_ user-facing change?
No API or configuration change. Concurrent stale writes now fail with an
optimistic lock exception, and policy overwrite retains version history.
### How was this patch tested?
- `./gradlew :core:spotlessApply`
- `env dockerTest=false ./gradlew :core:test --tests
org.apache.gravitino.storage.relational.service.TestPolicyMetaService
--tests
org.apache.gravitino.storage.relational.service.TestPolicyTagRelService
--tests
org.apache.gravitino.storage.relational.service.TestTagMetaService
-PskipDockerTests=true`
New tests: a metadata-only alter writing a complete snapshot, an
overwrite advancing the version while keeping history, an alter conflict
leaving no orphan version, a stale delete rolling back the relationship
cleanup, parent-metalake fencing on create, an overwrite landing on a
name held by another row, and the full cascade cleanup a delete
performs.
Only H2 runs locally; MySQL and PostgreSQL coverage comes from CI.
---
.../relational/mapper/PolicyMetaMapper.java | 96 +++-
.../mapper/PolicyMetaSQLProviderFactory.java | 28 +-
.../mapper/PolicyMetadataObjectRelMapper.java | 11 +-
.../PolicyMetadataObjectRelSQLProviderFactory.java | 8 +-
.../relational/mapper/PolicyTagRelMapper.java | 9 +
.../mapper/PolicyTagRelSQLProviderFactory.java | 5 +
.../relational/mapper/PolicyVersionMapper.java | 17 +-
.../mapper/PolicyVersionSQLProviderFactory.java | 11 +-
.../provider/base/PolicyMetaBaseSQLProvider.java | 92 ++--
.../PolicyMetadataObjectRelBaseSQLProvider.java | 16 +-
.../provider/base/PolicyTagRelBaseSQLProvider.java | 9 +
.../base/PolicyVersionBaseSQLProvider.java | 41 +-
.../postgresql/PolicyMetaPostgreSQLProvider.java | 33 +-
.../PolicyMetadataObjectRelPostgreSQLProvider.java | 12 +-
.../PolicyVersionPostgreSQLProvider.java | 38 +-
.../relational/service/OccWriteSupport.java | 14 +
.../relational/service/PolicyMetaService.java | 587 +++++++++++++++------
.../relational/service/PolicyTagRelService.java | 14 +-
.../storage/relational/utils/POConverters.java | 129 +++--
.../relational/service/TestOccWriteSupport.java | 17 +
.../relational/service/TestPolicyMetaService.java | 559 +++++++++++++++++++-
21 files changed, 1311 insertions(+), 435 deletions(-)
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyMetaMapper.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyMetaMapper.java
index 9862c5f151..eca19cb7fb 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyMetaMapper.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyMetaMapper.java
@@ -25,12 +25,25 @@ import org.apache.ibatis.annotations.InsertProvider;
import org.apache.ibatis.annotations.Param;
import org.apache.ibatis.annotations.Result;
import org.apache.ibatis.annotations.Results;
+import org.apache.ibatis.annotations.Select;
import org.apache.ibatis.annotations.SelectProvider;
import org.apache.ibatis.annotations.UpdateProvider;
public interface PolicyMetaMapper {
String POLICY_META_TABLE_NAME = "policy_meta";
+ /**
+ * Checks whether a soft-deleted policy still owns the requested primary key.
+ *
+ * @param policyId the policy ID
+ * @return one if a deleted row reserves the ID, otherwise zero
+ */
+ @Select(
+ "SELECT COUNT(*) FROM "
+ + POLICY_META_TABLE_NAME
+ + " WHERE policy_id = #{policyId} AND deleted_at > 0")
+ int countDeletedPolicyMetasById(@Param("policyId") Long policyId);
+
@Results({
@Result(property = "policyId", column = "policy_id"),
@Result(property = "policyName", column = "policy_name"),
@@ -76,11 +89,6 @@ public interface PolicyMetaMapper {
List<PolicyPO> listPolicyPOsByMetalakeAndPolicyNames(
@Param("metalakeName") String metalakeName, @Param("policyNames")
List<String> policyNames);
- @InsertProvider(
- type = PolicyMetaSQLProviderFactory.class,
- method = "insertPolicyMetaOnDuplicateKeyUpdate")
- void insertPolicyMetaOnDuplicateKeyUpdate(@Param("policyMeta") PolicyPO
policyPO);
-
@InsertProvider(type = PolicyMetaSQLProviderFactory.class, method =
"insertPolicyMeta")
void insertPolicyMeta(@Param("policyMeta") PolicyPO policyPO);
@@ -113,11 +121,18 @@ public interface PolicyMetaMapper {
@Param("newPolicyMeta") PolicyPO newPolicyMeta,
@Param("oldPolicyMeta") PolicyPO oldPolicyMeta);
+ /**
+ * Soft-deletes an active policy when its OCC version still matches.
+ *
+ * @param policyId The policy ID.
+ * @param currentVersion The version observed by the caller.
+ * @return The number of affected rows.
+ */
@UpdateProvider(
type = PolicyMetaSQLProviderFactory.class,
- method = "softDeletePolicyByMetalakeAndPolicyName")
- Integer softDeletePolicyByMetalakeAndPolicyName(
- @Param("metalakeName") String metalakeName, @Param("policyName") String
policyName);
+ method = "softDeletePolicyByIdAndVersion")
+ Integer softDeletePolicyByIdAndVersion(
+ @Param("policyId") Long policyId, @Param("currentVersion") Long
currentVersion);
@UpdateProvider(
type = PolicyMetaSQLProviderFactory.class,
@@ -146,6 +161,29 @@ public interface PolicyMetaMapper {
PolicyPO selectPolicyMetaByMetalakeIdAndName(
@Param("metalakeId") long metalakeId, @Param("policyName") String
policyName);
+ /**
+ * Selects and exclusively locks an active policy by its natural key.
+ *
+ * @param metalakeId The metalake ID.
+ * @param policyName The policy name.
+ * @return The locked policy, or null if the natural key is not active.
+ */
+ @Results({
+ @Result(property = "policyId", column = "policy_id"),
+ @Result(property = "policyName", column = "policy_name"),
+ @Result(property = "policyType", column = "policy_type"),
+ @Result(property = "metalakeId", column = "metalake_id"),
+ @Result(property = "auditInfo", column = "audit_info"),
+ @Result(property = "currentVersion", column = "current_version"),
+ @Result(property = "lastVersion", column = "last_version"),
+ @Result(property = "deletedAt", column = "deleted_at")
+ })
+ @SelectProvider(
+ type = PolicyMetaSQLProviderFactory.class,
+ method = "selectPolicyMetaByMetalakeIdAndNameForUpdate")
+ PolicyPO selectPolicyMetaByMetalakeIdAndNameForUpdate(
+ @Param("metalakeId") long metalakeId, @Param("policyName") String
policyName);
+
@Results({
@Result(property = "policyId", column = "policy_id"),
@Result(property = "policyName", column = "policy_name"),
@@ -159,6 +197,27 @@ public interface PolicyMetaMapper {
@SelectProvider(type = PolicyMetaSQLProviderFactory.class, method =
"selectPolicyByPolicyId")
PolicyPO selectPolicyByPolicyId(@Param("policyId") Long policyId);
+ /**
+ * Selects and exclusively locks an active policy by ID.
+ *
+ * @param policyId The policy ID.
+ * @return The locked policy, or null if it is not active.
+ */
+ @Results({
+ @Result(property = "policyId", column = "policy_id"),
+ @Result(property = "policyName", column = "policy_name"),
+ @Result(property = "policyType", column = "policy_type"),
+ @Result(property = "metalakeId", column = "metalake_id"),
+ @Result(property = "auditInfo", column = "audit_info"),
+ @Result(property = "currentVersion", column = "current_version"),
+ @Result(property = "lastVersion", column = "last_version"),
+ @Result(property = "deletedAt", column = "deleted_at")
+ })
+ @SelectProvider(
+ type = PolicyMetaSQLProviderFactory.class,
+ method = "selectPolicyByPolicyIdForUpdate")
+ PolicyPO selectPolicyByPolicyIdForUpdate(@Param("policyId") Long policyId);
+
@Results({
@Result(property = "policyId", column = "policy_id"),
@Result(property = "policyName", column = "policy_name"),
@@ -172,6 +231,27 @@ public interface PolicyMetaMapper {
@SelectProvider(type = PolicyMetaSQLProviderFactory.class, method =
"listPolicyPOsByPolicyIds")
List<PolicyPO> listPolicyPOsByPolicyIds(@Param("policyIds") List<Long>
policyIds);
+ /**
+ * Selects and exclusively locks the active policies with the given IDs, in
ascending ID order.
+ *
+ * @param policyIds The policy IDs to lock.
+ * @return The locked policies. Policies that are not active are absent from
the result.
+ */
+ @Results({
+ @Result(property = "policyId", column = "policy_id"),
+ @Result(property = "policyName", column = "policy_name"),
+ @Result(property = "policyType", column = "policy_type"),
+ @Result(property = "metalakeId", column = "metalake_id"),
+ @Result(property = "auditInfo", column = "audit_info"),
+ @Result(property = "currentVersion", column = "current_version"),
+ @Result(property = "lastVersion", column = "last_version"),
+ @Result(property = "deletedAt", column = "deleted_at")
+ })
+ @SelectProvider(
+ type = PolicyMetaSQLProviderFactory.class,
+ method = "listPolicyPOsByPolicyIdsForUpdate")
+ List<PolicyPO> listPolicyPOsByPolicyIdsForUpdate(@Param("policyIds")
List<Long> policyIds);
+
@Results({
@Result(property = "policyId", column = "policy_id"),
@Result(property = "policyName", column = "policy_name"),
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyMetaSQLProviderFactory.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyMetaSQLProviderFactory.java
index 4f7b1d1177..807badd1fb 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyMetaSQLProviderFactory.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyMetaSQLProviderFactory.java
@@ -57,11 +57,6 @@ public class PolicyMetaSQLProviderFactory {
return getProvider().listPolicyPOsByMetalakeAndPolicyNames(metalakeName,
policyNames);
}
- public static String insertPolicyMetaOnDuplicateKeyUpdate(
- @Param("policyMeta") PolicyPO policyPO) {
- return getProvider().insertPolicyMetaOnDuplicateKeyUpdate(policyPO);
- }
-
public static String insertPolicyMeta(@Param("policyMeta") PolicyPO
policyPO) {
return getProvider().insertPolicyMeta(policyPO);
}
@@ -77,9 +72,10 @@ public class PolicyMetaSQLProviderFactory {
return getProvider().updatePolicyMeta(newPolicyMeta, oldPolicyMeta);
}
- public static String softDeletePolicyByMetalakeAndPolicyName(
- @Param("metalakeName") String metalakeName, @Param("policyName") String
policyName) {
- return getProvider().softDeletePolicyByMetalakeAndPolicyName(metalakeName,
policyName);
+ /** Delegates a version-checked policy soft delete. */
+ public static String softDeletePolicyByIdAndVersion(
+ @Param("policyId") Long policyId, @Param("currentVersion") Long
currentVersion) {
+ return getProvider().softDeletePolicyByIdAndVersion(policyId,
currentVersion);
}
public static String deletePolicyMetasByLegacyTimeline(
@@ -96,10 +92,26 @@ public class PolicyMetaSQLProviderFactory {
return getProvider().selectPolicyMetaByMetalakeIdAndName(metalakeId,
policyName);
}
+ /** Delegates an exclusive-lock policy query by natural key. */
+ public static String selectPolicyMetaByMetalakeIdAndNameForUpdate(
+ @Param("metalakeId") Long metalakeId, @Param("policyName") String
policyName) {
+ return
getProvider().selectPolicyMetaByMetalakeIdAndNameForUpdate(metalakeId,
policyName);
+ }
+
public static String selectPolicyByPolicyId(@Param("policyId") Long
policyId) {
return getProvider().selectPolicyByPolicyId(policyId);
}
+ /** Delegates an exclusive-lock policy query. */
+ public static String selectPolicyByPolicyIdForUpdate(@Param("policyId") Long
policyId) {
+ return getProvider().selectPolicyByPolicyIdForUpdate(policyId);
+ }
+
+ /** Delegates a locking read of several policies. */
+ public static String listPolicyPOsByPolicyIdsForUpdate(@Param("policyIds")
List<Long> policyIds) {
+ return getProvider().listPolicyPOsByPolicyIdsForUpdate(policyIds);
+ }
+
public static String listPolicyPOsByPolicyIds(@Param("policyIds") List<Long>
policyIds) {
return getProvider().listPolicyPOsByPolicyIds(policyIds);
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyMetadataObjectRelMapper.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyMetadataObjectRelMapper.java
index 31226a0eeb..b3ddb04368 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyMetadataObjectRelMapper.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyMetadataObjectRelMapper.java
@@ -103,11 +103,16 @@ public interface PolicyMetadataObjectRelMapper {
@Param("metadataObjectType") String metadataObjectType,
@Param("policyIds") List<Long> policyIds);
+ /**
+ * Soft-deletes all active metadata-object relations for a policy.
+ *
+ * @param policyId The policy ID.
+ * @return The number of affected rows.
+ */
@UpdateProvider(
type = PolicyMetadataObjectRelSQLProviderFactory.class,
- method = "softDeletePolicyMetadataObjectRelsByMetalakeAndPolicyName")
- Integer softDeletePolicyMetadataObjectRelsByMetalakeAndPolicyName(
- @Param("metalakeName") String metalakeName, @Param("policyName") String
policyName);
+ method = "softDeletePolicyMetadataObjectRelsByPolicyId")
+ Integer softDeletePolicyMetadataObjectRelsByPolicyId(@Param("policyId") Long
policyId);
@UpdateProvider(
type = PolicyMetadataObjectRelSQLProviderFactory.class,
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyMetadataObjectRelSQLProviderFactory.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyMetadataObjectRelSQLProviderFactory.java
index 8b38bfb3e0..e968bfcc44 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyMetadataObjectRelSQLProviderFactory.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyMetadataObjectRelSQLProviderFactory.java
@@ -89,10 +89,10 @@ public class PolicyMetadataObjectRelSQLProviderFactory {
metadataObjectId, metadataObjectType, policyIds);
}
- public static String
softDeletePolicyMetadataObjectRelsByMetalakeAndPolicyName(
- @Param("metalakeName") String metalakeName, @Param("policyName") String
policyName) {
- return getProvider()
-
.softDeletePolicyMetadataObjectRelsByMetalakeAndPolicyName(metalakeName,
policyName);
+ /** Delegates cleanup of metadata-object relations by policy ID. */
+ public static String softDeletePolicyMetadataObjectRelsByPolicyId(
+ @Param("policyId") Long policyId) {
+ return
getProvider().softDeletePolicyMetadataObjectRelsByPolicyId(policyId);
}
public static String softDeletePolicyMetadataObjectRelsByMetalakeId(
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 35a97b04a8..f1b61a0c43 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 tag relation for a policy.
+ *
+ * @param policyId The policy ID.
+ * @return The number of affected rows.
+ */
+ @UpdateProvider(type = PolicyTagRelSQLProviderFactory.class, method =
"softDeleteByPolicyId")
+ int softDeleteByPolicyId(@Param("policyId") Long policyId);
+
/**
* Soft-deletes every active policy relation for a tag.
*
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 5a82d68d77..83667a9c7a 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 policy deletion cleanup. */
+ public static String softDeleteByPolicyId(@Param("policyId") Long policyId) {
+ return getProvider().softDeleteByPolicyId(policyId);
+ }
+
/** Delegates tag deletion cleanup. */
public static String softDeleteByTagId(@Param("tagId") Long tagId) {
return getProvider().softDeleteByTagId(tagId);
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyVersionMapper.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyVersionMapper.java
index 9bfbc62ea7..c926d60bac 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyVersionMapper.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyVersionMapper.java
@@ -29,20 +29,19 @@ import org.apache.ibatis.annotations.UpdateProvider;
public interface PolicyVersionMapper {
String POLICY_VERSION_TABLE_NAME = "policy_version_info";
- @InsertProvider(
- type = PolicyVersionSQLProviderFactory.class,
- method = "insertPolicyVersionOnDuplicateKeyUpdate")
- void insertPolicyVersionOnDuplicateKeyUpdate(
- @Param("policyVersion") PolicyVersionPO policyVersionPO);
-
@InsertProvider(type = PolicyVersionSQLProviderFactory.class, method =
"insertPolicyVersion")
void insertPolicyVersion(@Param("policyVersion") PolicyVersionPO
policyVersionPO);
+ /**
+ * Soft-deletes all active content snapshots for a policy.
+ *
+ * @param policyId The policy ID.
+ * @return The number of affected rows.
+ */
@UpdateProvider(
type = PolicyVersionSQLProviderFactory.class,
- method = "softDeletePolicyVersionByMetalakeAndPolicyName")
- Integer softDeletePolicyVersionByMetalakeAndPolicyName(
- @Param("metalakeName") String metalakeName, @Param("policyName") String
policyName);
+ method = "softDeletePolicyVersionsByPolicyId")
+ Integer softDeletePolicyVersionsByPolicyId(@Param("policyId") Long policyId);
@UpdateProvider(
type = PolicyVersionSQLProviderFactory.class,
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyVersionSQLProviderFactory.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyVersionSQLProviderFactory.java
index 9fbe4ddc79..3e71ebf771 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyVersionSQLProviderFactory.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyVersionSQLProviderFactory.java
@@ -47,19 +47,14 @@ public class PolicyVersionSQLProviderFactory {
return POLICY_VERSION_SQL_PROVIDER_MAP.get(jdbcBackendType);
}
- public static String insertPolicyVersionOnDuplicateKeyUpdate(
- @Param("policyVersion") PolicyVersionPO policyVersionPO) {
- return
getProvider().insertPolicyVersionOnDuplicateKeyUpdate(policyVersionPO);
- }
-
public static String insertPolicyVersion(
@Param("policyVersion") PolicyVersionPO policyVersionPO) {
return getProvider().insertPolicyVersion(policyVersionPO);
}
- public static String softDeletePolicyVersionByMetalakeAndPolicyName(
- @Param("metalakeName") String metalakeName, @Param("policyName") String
policyName) {
- return
getProvider().softDeletePolicyVersionByMetalakeAndPolicyName(metalakeName,
policyName);
+ /** Delegates cleanup of policy snapshots by policy ID. */
+ public static String softDeletePolicyVersionsByPolicyId(@Param("policyId")
Long policyId) {
+ return getProvider().softDeletePolicyVersionsByPolicyId(policyId);
}
public static String deletePolicyVersionsByLegacyTimeline(
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/PolicyMetaBaseSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/PolicyMetaBaseSQLProvider.java
index 909b9fe246..730ce9bed0 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/PolicyMetaBaseSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/PolicyMetaBaseSQLProvider.java
@@ -74,24 +74,6 @@ public class PolicyMetaBaseSQLProvider {
+ "</script>";
}
- public String insertPolicyMetaOnDuplicateKeyUpdate(@Param("policyMeta")
PolicyPO policyPO) {
- return "INSERT INTO "
- + POLICY_META_TABLE_NAME
- + " (policy_id, policy_name, policy_type, metalake_id,"
- + " audit_info, current_version, last_version, deleted_at)"
- + " VALUES (#{policyMeta.policyId}, #{policyMeta.policyName},
#{policyMeta.policyType},"
- + " #{policyMeta.metalakeId}, #{policyMeta.auditInfo},
#{policyMeta.currentVersion},"
- + " #{policyMeta.lastVersion}, #{policyMeta.deletedAt})"
- + " ON DUPLICATE KEY UPDATE"
- + " policy_name = #{policyMeta.policyName},"
- + " policy_type = #{policyMeta.policyType},"
- + " metalake_id = #{policyMeta.metalakeId},"
- + " audit_info = #{policyMeta.auditInfo},"
- + " current_version = #{policyMeta.currentVersion},"
- + " last_version = #{policyMeta.lastVersion},"
- + " deleted_at = #{policyMeta.deletedAt}";
- }
-
public String insertPolicyMeta(@Param("policyMeta") PolicyPO policyPO) {
return "INSERT INTO "
+ POLICY_META_TABLE_NAME
@@ -136,26 +118,19 @@ public class PolicyMetaBaseSQLProvider {
+ " last_version = #{newPolicyMeta.lastVersion},"
+ " deleted_at = #{newPolicyMeta.deletedAt}"
+ " WHERE policy_id = #{oldPolicyMeta.policyId}"
- + " AND policy_name = #{oldPolicyMeta.policyName}"
- + " AND policy_type = #{oldPolicyMeta.policyType}"
- + " AND metalake_id = #{oldPolicyMeta.metalakeId}"
- + " AND audit_info = #{oldPolicyMeta.auditInfo}"
+ " AND current_version = #{oldPolicyMeta.currentVersion}"
- + " AND last_version = #{oldPolicyMeta.lastVersion}"
+ " AND deleted_at = 0";
}
- public String softDeletePolicyByMetalakeAndPolicyName(
- @Param("metalakeName") String metalakeName, @Param("policyName") String
policyName) {
+ /** Returns SQL that soft-deletes a policy using its stable ID and observed
OCC version. */
+ public String softDeletePolicyByIdAndVersion(
+ @Param("policyId") Long policyId, @Param("currentVersion") Long
currentVersion) {
return "UPDATE "
+ POLICY_META_TABLE_NAME
- + " pm SET pm.deleted_at = "
+ + " SET deleted_at = "
+ DatabaseTimeSQL.MYSQL
- + " WHERE pm.metalake_id IN ("
- + " SELECT mm.metalake_id FROM "
- + MetalakeMetaMapper.TABLE_NAME
- + " mm WHERE mm.metalake_name = #{metalakeName} AND mm.deleted_at = 0)"
- + " AND pm.policy_name = #{policyName} AND pm.deleted_at = 0";
+ + " WHERE policy_id = #{policyId} AND current_version =
#{currentVersion}"
+ + " AND deleted_at = 0";
}
public String deletePolicyMetasByLegacyTimeline(
@@ -185,21 +160,23 @@ public class PolicyMetaBaseSQLProvider {
+ " AND pm.deleted_at = 0 ";
}
- public String listPolicyPOsByPolicyIds(@Param("policyIds") List<Long>
policyIds) {
+ /** Returns SQL that selects and exclusively locks an active policy by ID. */
+ public String selectPolicyByPolicyIdForUpdate(@Param("policyId") Long
policyId) {
+ return selectPolicyByPolicyId(policyId) + " FOR UPDATE";
+ }
+
+ /**
+ * Returns SQL that selects and exclusively locks several active policies,
ordered by policy ID so
+ * that concurrent callers take the row locks in the same order.
+ */
+ public String listPolicyPOsByPolicyIdsForUpdate(@Param("policyIds")
List<Long> policyIds) {
return "<script>"
- + "SELECT pm.policy_id, pm.policy_name, pm.policy_type,
pm.metalake_id,"
- + " pm.audit_info, pm.current_version, pm.last_version,"
- + " pm.deleted_at"
- + " FROM "
- + POLICY_META_TABLE_NAME
- + " pm"
- + " WHERE pm.deleted_at = 0"
- + " AND pm.policy_id IN ("
- + "<foreach collection='policyIds' item='policyId' separator=','>"
- + "#{policyId}"
- + "</foreach>"
- + ")"
- + "</script>";
+ + selectPolicyPOsByPolicyIdsBody()
+ + " ORDER BY pm.policy_id FOR UPDATE</script>";
+ }
+
+ public String listPolicyPOsByPolicyIds(@Param("policyIds") List<Long>
policyIds) {
+ return "<script>" + selectPolicyPOsByPolicyIdsBody() + "</script>";
}
public String selectPolicyMetaByMetalakeIdAndName(
@@ -216,6 +193,12 @@ public class PolicyMetaBaseSQLProvider {
+ " AND pm.deleted_at = 0 ";
}
+ /** Returns SQL that selects and exclusively locks an active policy by its
natural key. */
+ public String selectPolicyMetaByMetalakeIdAndNameForUpdate(
+ @Param("metalakeId") Long metalakeId, @Param("policyName") String
policyName) {
+ return selectPolicyMetaByMetalakeIdAndName(metalakeId, policyName) + " FOR
UPDATE";
+ }
+
public String batchSelectPolicyByIdentifier(
@Param("metalakeName") String metalakeName, @Param("policyNames")
List<String> policyNames) {
return "<script>"
@@ -241,4 +224,23 @@ public class PolicyMetaBaseSQLProvider {
+ " AND pm.deleted_at = 0 AND pv.deleted_at = 0 AND mm.deleted_at = 0"
+ "</script>";
}
+
+ /**
+ * Returns the shared body of the by-ID list queries, without the enclosing
{@code <script>} tag,
+ * so the plain and the locking variant cannot drift apart when the selected
columns change.
+ */
+ private String selectPolicyPOsByPolicyIdsBody() {
+ return "SELECT pm.policy_id, pm.policy_name, pm.policy_type,
pm.metalake_id,"
+ + " pm.audit_info, pm.current_version, pm.last_version,"
+ + " pm.deleted_at"
+ + " FROM "
+ + POLICY_META_TABLE_NAME
+ + " pm"
+ + " WHERE pm.deleted_at = 0"
+ + " AND pm.policy_id IN ("
+ + "<foreach collection='policyIds' item='policyId' separator=','>"
+ + "#{policyId}"
+ + "</foreach>"
+ + ")";
+ }
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/PolicyMetadataObjectRelBaseSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/PolicyMetadataObjectRelBaseSQLProvider.java
index dbd444b0c3..e239139d5a 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/PolicyMetadataObjectRelBaseSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/PolicyMetadataObjectRelBaseSQLProvider.java
@@ -134,19 +134,13 @@ public class PolicyMetadataObjectRelBaseSQLProvider {
+ "</script>";
}
- public String softDeletePolicyMetadataObjectRelsByMetalakeAndPolicyName(
- @Param("metalakeName") String metalakeName, @Param("policyName") String
policyName) {
+ /** Returns SQL that soft-deletes every active metadata-object relation for
a policy ID. */
+ public String
softDeletePolicyMetadataObjectRelsByPolicyId(@Param("policyId") Long policyId) {
return "UPDATE "
+
PolicyMetadataObjectRelMapper.POLICY_METADATA_OBJECT_RELATION_TABLE_NAME
- + " pe JOIN "
- + PolicyMetaMapper.POLICY_META_TABLE_NAME
- + " pm ON pe.policy_id = pm.policy_id JOIN "
- + MetalakeMetaMapper.TABLE_NAME
- + " mm ON pm.metalake_id = mm.metalake_id"
- + " SET pe.deleted_at = "
- + DatabaseTimeSQL.MYSQL
- + " WHERE mm.metalake_name = #{metalakeName} AND pm.policy_name =
#{policyName}"
- + " AND pe.deleted_at = 0 AND pm.deleted_at = 0 AND mm.deleted_at = 0";
+ + " SET deleted_at = (UNIX_TIMESTAMP() * 1000.0)"
+ + " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000"
+ + " WHERE policy_id = #{policyId} AND deleted_at = 0";
}
public String softDeletePolicyMetadataObjectRelsByMetalakeId(
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 826e37ba88..b193b1e01b 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 tag relations when a policy is deleted. */
+ public String softDeleteByPolicyId(@Param("policyId") Long policyId) {
+ return "UPDATE "
+ + POLICY_TAG_RELATION_TABLE_NAME
+ + " SET deleted_at = "
+ + deletedAtNowExpression()
+ + " WHERE policy_id = #{policyId} 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 "
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/PolicyVersionBaseSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/PolicyVersionBaseSQLProvider.java
index 7912253daa..3db120be11 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/PolicyVersionBaseSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/PolicyVersionBaseSQLProvider.java
@@ -18,33 +18,14 @@
*/
package org.apache.gravitino.storage.relational.mapper.provider.base;
-import static
org.apache.gravitino.storage.relational.mapper.PolicyMetaMapper.POLICY_META_TABLE_NAME;
import static
org.apache.gravitino.storage.relational.mapper.PolicyVersionMapper.POLICY_VERSION_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.po.PolicyVersionPO;
import org.apache.ibatis.annotations.Param;
public class PolicyVersionBaseSQLProvider {
- public String insertPolicyVersionOnDuplicateKeyUpdate(
- @Param("policyVersion") PolicyVersionPO policyVersion) {
- return "INSERT INTO "
- + POLICY_VERSION_TABLE_NAME
- + " (metalake_id, policy_id, version, policy_comment, enabled,
content, deleted_at)"
- + " VALUES (#{policyVersion.metalakeId}, #{policyVersion.policyId},
#{policyVersion.version}, #{policyVersion.policyComment},"
- + " #{policyVersion.enabled}, #{policyVersion.content},
#{policyVersion.deletedAt})"
- + " ON DUPLICATE KEY UPDATE"
- + " metalake_id = #{policyVersion.metalakeId},"
- + " policy_id = #{policyVersion.policyId},"
- + " version = #{policyVersion.version},"
- + " policy_comment = #{policyVersion.policyComment},"
- + " enabled = #{policyVersion.enabled},"
- + " content = #{policyVersion.content},"
- + " deleted_at = #{policyVersion.deletedAt}";
- }
-
public String insertPolicyVersion(@Param("policyVersion") PolicyVersionPO
policyVersion) {
return "INSERT INTO "
+ POLICY_VERSION_TABLE_NAME
@@ -54,25 +35,13 @@ public class PolicyVersionBaseSQLProvider {
+ " #{policyVersion.deletedAt})";
}
- public String softDeletePolicyVersionByMetalakeAndPolicyName(
- @Param("metalakeName") String metalakeName, @Param("policyName") String
policyName) {
+ /** Returns SQL that soft-deletes every active content snapshot for a policy
ID. */
+ public String softDeletePolicyVersionsByPolicyId(@Param("policyId") Long
policyId) {
return "UPDATE "
+ POLICY_VERSION_TABLE_NAME
- + " pv SET pv.deleted_at = "
- + DatabaseTimeSQL.MYSQL
- + " WHERE pv.metalake_id IN ("
- + " SELECT mm.metalake_id FROM "
- + MetalakeMetaMapper.TABLE_NAME
- + " mm WHERE mm.metalake_name = #{metalakeName} AND mm.deleted_at = 0)"
- + " AND pv.policy_id IN ("
- + " SELECT pm.policy_id FROM "
- + POLICY_META_TABLE_NAME
- + " pm WHERE pm.policy_name = #{policyName} AND pm.deleted_at = 0"
- + " AND pm.metalake_id IN ("
- + " SELECT mm.metalake_id FROM "
- + MetalakeMetaMapper.TABLE_NAME
- + " mm WHERE mm.metalake_name = #{metalakeName} AND mm.deleted_at =
0))"
- + " AND pv.deleted_at = 0";
+ + " SET deleted_at = (UNIX_TIMESTAMP() * 1000.0)"
+ + " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000"
+ + " WHERE policy_id = #{policyId} AND deleted_at = 0";
}
public String deletePolicyVersionsByLegacyTimeline(
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/PolicyMetaPostgreSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/PolicyMetaPostgreSQLProvider.java
index 1302ca9774..e2c5770da2 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/PolicyMetaPostgreSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/PolicyMetaPostgreSQLProvider.java
@@ -22,19 +22,17 @@ import static
org.apache.gravitino.storage.relational.mapper.PolicyMetaMapper.PO
import org.apache.gravitino.storage.relational.mapper.provider.DatabaseTimeSQL;
import
org.apache.gravitino.storage.relational.mapper.provider.base.PolicyMetaBaseSQLProvider;
-import org.apache.gravitino.storage.relational.po.PolicyPO;
public class PolicyMetaPostgreSQLProvider extends PolicyMetaBaseSQLProvider {
@Override
- public String softDeletePolicyByMetalakeAndPolicyName(String metalakeName,
String policyName) {
+ public String softDeletePolicyByIdAndVersion(Long policyId, Long
currentVersion) {
return "UPDATE "
+ POLICY_META_TABLE_NAME
+ " SET deleted_at = "
+ DatabaseTimeSQL.POSTGRESQL
- + " WHERE metalake_id = (SELECT metalake_id FROM "
- + " metalake_meta mm WHERE mm.metalake_name = #{metalakeName} AND
mm.deleted_at = 0)"
- + " AND policy_name = #{policyName} AND deleted_at = 0";
+ + " WHERE policy_id = #{policyId} AND current_version =
#{currentVersion}"
+ + " AND deleted_at = 0";
}
@Override
@@ -54,29 +52,4 @@ public class PolicyMetaPostgreSQLProvider extends
PolicyMetaBaseSQLProvider {
+ POLICY_META_TABLE_NAME
+ " WHERE deleted_at > 0 AND deleted_at < #{legacyTimeline} LIMIT
#{limit})";
}
-
- @Override
- public String insertPolicyMetaOnDuplicateKeyUpdate(PolicyPO policyPO) {
- return "INSERT INTO "
- + POLICY_META_TABLE_NAME
- + " (policy_id, policy_name, policy_type, metalake_id,"
- + " audit_info, current_version, last_version, deleted_at)"
- + " VALUES ("
- + " #{policyMeta.policyId},"
- + " #{policyMeta.policyName},"
- + " #{policyMeta.policyType},"
- + " #{policyMeta.metalakeId},"
- + " #{policyMeta.auditInfo},"
- + " #{policyMeta.currentVersion},"
- + " #{policyMeta.lastVersion},"
- + " #{policyMeta.deletedAt})"
- + " ON CONFLICT (policy_id) DO UPDATE SET"
- + " policy_name = #{policyMeta.policyName},"
- + " policy_type = #{policyMeta.policyType},"
- + " metalake_id = #{policyMeta.metalakeId},"
- + " audit_info = #{policyMeta.auditInfo},"
- + " current_version = #{policyMeta.currentVersion},"
- + " last_version = #{policyMeta.lastVersion},"
- + " deleted_at = #{policyMeta.deletedAt}";
- }
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/PolicyMetadataObjectRelPostgreSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/PolicyMetadataObjectRelPostgreSQLProvider.java
index 0978873a1e..e2ae7b9345 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/PolicyMetadataObjectRelPostgreSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/PolicyMetadataObjectRelPostgreSQLProvider.java
@@ -41,18 +41,12 @@ public class PolicyMetadataObjectRelPostgreSQLProvider
private static final String DELETED_AT_NOW_EXPRESSION = " " +
DatabaseTimeSQL.POSTGRESQL;
@Override
- public String softDeletePolicyMetadataObjectRelsByMetalakeAndPolicyName(
- String metalakeName, String policyName) {
+ public String softDeletePolicyMetadataObjectRelsByPolicyId(Long policyId) {
return "UPDATE "
+ POLICY_METADATA_OBJECT_RELATION_TABLE_NAME
- + " te SET deleted_at ="
+ + " SET deleted_at ="
+ DELETED_AT_NOW_EXPRESSION
- + " WHERE te.policy_id IN (SELECT tm.policy_id FROM "
- + PolicyMetaMapper.POLICY_META_TABLE_NAME
- + " tm 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.policy_name = #{policyName} AND tm.deleted_at = 0) AND
te.deleted_at = 0";
+ + " WHERE policy_id = #{policyId} AND deleted_at = 0";
}
@Override
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/PolicyVersionPostgreSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/PolicyVersionPostgreSQLProvider.java
index e0c7b9f0a1..7042f1f1e5 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/PolicyVersionPostgreSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/PolicyVersionPostgreSQLProvider.java
@@ -18,29 +18,18 @@
*/
package org.apache.gravitino.storage.relational.mapper.provider.postgresql;
-import static
org.apache.gravitino.storage.relational.mapper.PolicyMetaMapper.POLICY_META_TABLE_NAME;
import static
org.apache.gravitino.storage.relational.mapper.PolicyVersionMapper.POLICY_VERSION_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.PolicyVersionBaseSQLProvider;
-import org.apache.gravitino.storage.relational.po.PolicyVersionPO;
public class PolicyVersionPostgreSQLProvider extends
PolicyVersionBaseSQLProvider {
@Override
- public String softDeletePolicyVersionByMetalakeAndPolicyName(
- String metalakeName, String policyName) {
+ public String softDeletePolicyVersionsByPolicyId(Long policyId) {
return "UPDATE "
+ POLICY_VERSION_TABLE_NAME
- + " SET deleted_at = "
- + DatabaseTimeSQL.POSTGRESQL
- + " WHERE metalake_id = (SELECT metalake_id FROM "
- + MetalakeMetaMapper.TABLE_NAME
- + " mm WHERE mm.metalake_name = #{metalakeName} AND mm.deleted_at = 0)"
- + " AND policy_id = (SELECT policy_id FROM "
- + POLICY_META_TABLE_NAME
- + " pm WHERE pm.policy_name = #{policyName} AND pm.deleted_at = 0)"
- + " AND deleted_at = 0";
+ + " SET deleted_at = CAST(EXTRACT(EPOCH FROM CURRENT_TIMESTAMP) * 1000
AS BIGINT)"
+ + " WHERE policy_id = #{policyId} AND deleted_at = 0";
}
@Override
@@ -73,25 +62,4 @@ public class PolicyVersionPostgreSQLProvider extends
PolicyVersionBaseSQLProvide
+ DatabaseTimeSQL.POSTGRESQL
+ " WHERE metalake_id = #{metalakeId} AND deleted_at = 0";
}
-
- @Override
- public String insertPolicyVersionOnDuplicateKeyUpdate(PolicyVersionPO
policyVersion) {
- return "INSERT INTO "
- + POLICY_VERSION_TABLE_NAME
- + " (metalake_id, policy_id, version, policy_comment, enabled,"
- + " content, deleted_at)"
- + " VALUES ("
- + " #{policyVersion.metalakeId},"
- + " #{policyVersion.policyId},"
- + " #{policyVersion.version},"
- + " #{policyVersion.policyComment},"
- + " #{policyVersion.enabled},"
- + " #{policyVersion.content},"
- + " #{policyVersion.deletedAt})"
- + " ON CONFLICT (policy_id, version, deleted_at) DO UPDATE SET"
- + " policy_comment = #{policyVersion.policyComment},"
- + " enabled = #{policyVersion.enabled},"
- + " content = #{policyVersion.content},"
- + " deleted_at = #{policyVersion.deletedAt}";
- }
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/OccWriteSupport.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/OccWriteSupport.java
index f9797e0646..01842bfe19 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/service/OccWriteSupport.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/OccWriteSupport.java
@@ -115,6 +115,20 @@ public class OccWriteSupport {
}
}
+ /**
+ * Executes a single-row compare-and-set update for an entity guarded by
version.
+ *
+ * @param updateOps an operation supplying the number of rows affected by
the update
+ * @param onMissSupplier a supplier providing the RuntimeException when zero
rows are updated
+ */
+ public static void updateWithVersion(
+ IntSupplier updateOps, Supplier<RuntimeException> onMissSupplier) {
+ int updated = updateOps.getAsInt();
+ if (updated == 0) {
+ throw onMissSupplier.get();
+ }
+ }
+
/**
* Executes a batch soft delete of child entities guarded by their
individual versions.
*
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/PolicyMetaService.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/PolicyMetaService.java
index 0c623f96c0..71ed987f9c 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/service/PolicyMetaService.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/PolicyMetaService.java
@@ -20,11 +20,15 @@ package org.apache.gravitino.storage.relational.service;
import static
org.apache.gravitino.metrics.source.MetricsSource.GRAVITINO_RELATIONAL_STORE_METRIC_NAME;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Map;
import java.util.Objects;
import java.util.function.Function;
import java.util.stream.Collectors;
@@ -38,9 +42,15 @@ import org.apache.gravitino.exceptions.NoSuchEntityException;
import org.apache.gravitino.meta.GenericEntity;
import org.apache.gravitino.meta.PolicyEntity;
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.PolicyMetaMapper;
import
org.apache.gravitino.storage.relational.mapper.PolicyMetadataObjectRelMapper;
+import org.apache.gravitino.storage.relational.mapper.PolicyTagRelMapper;
import org.apache.gravitino.storage.relational.mapper.PolicyVersionMapper;
+import org.apache.gravitino.storage.relational.mapper.SecurableObjectMapper;
+import
org.apache.gravitino.storage.relational.mapper.TagMetadataObjectRelMapper;
+import org.apache.gravitino.storage.relational.po.MetalakePO;
import org.apache.gravitino.storage.relational.po.PolicyMaxVersionPO;
import org.apache.gravitino.storage.relational.po.PolicyMetadataObjectRelPO;
import org.apache.gravitino.storage.relational.po.PolicyPO;
@@ -93,36 +103,35 @@ public class PolicyMetaService {
String metalakeName = ns.level(0);
try {
- Long metalakeId =
- EntityIdService.getEntityId(NameIdentifier.of(metalakeName),
Entity.EntityType.METALAKE);
+ 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);
+ }
- PolicyPO.Builder builder = PolicyPO.builder().withMetalakeId(metalakeId);
+ PolicyPO.Builder builder =
PolicyPO.builder().withMetalakeId(metalakePO.getMetalakeId());
PolicyPO policyPO =
POConverters.initializePolicyPOWithVersion(policyEntity, builder);
- // insert both policy meta table and policy version table
SessionUtils.doMultipleWithCommit(
- () ->
- SessionUtils.doWithoutCommit(
- PolicyMetaMapper.class,
- mapper -> {
- if (overwritten) {
- mapper.insertPolicyMetaOnDuplicateKeyUpdate(policyPO);
- } else {
- mapper.insertPolicyMeta(policyPO);
- }
- }),
- () ->
- SessionUtils.doWithoutCommit(
- PolicyVersionMapper.class,
- mapper -> {
- if (overwritten) {
-
mapper.insertPolicyVersionOnDuplicateKeyUpdate(policyPO.getPolicyVersionPO());
- } else {
-
mapper.insertPolicyVersion(policyPO.getPolicyVersionPO());
- }
- }));
+ () -> lockMetalakeForPolicyCreate(metalakePO),
+ () -> insertPolicyWithoutCommit(policyEntity, policyPO,
overwritten));
} catch (RuntimeException e) {
- ExceptionUtils.checkSQLException(e, Entity.EntityType.POLICY,
policyEntity.toString());
+ try {
+ ExceptionUtils.checkSQLException(
+ e, Entity.EntityType.POLICY,
policyEntity.nameIdentifier().toString());
+ } catch (EntityAlreadyExistsException duplicate) {
+ if (overwritten) {
+ // A missing-row locking read does not fence a concurrent insert at
READ_COMMITTED.
+ // Propagate the conflict so the whole transaction is rolled back
before retrying.
+ throw ExceptionUtils.concurrentModification(
+ Entity.EntityType.POLICY, policyEntity.nameIdentifier());
+ }
+ throw duplicate;
+ }
throw e;
}
}
@@ -143,69 +152,87 @@ public class PolicyMetaService {
updatedPolicyEntity.id(),
oldPolicyEntity.id());
- Integer updateResult;
try {
- boolean checkNeedUpdateVersion =
- POConverters.checkPolicyVersionNeedUpdate(
- oldPolicyPO.getPolicyVersionPO(), updatedPolicyEntity);
PolicyPO newPolicyPO =
- POConverters.updatePolicyPOWithVersion(
- oldPolicyPO, updatedPolicyEntity, checkNeedUpdateVersion);
- if (checkNeedUpdateVersion) {
- SessionUtils.doMultipleWithCommit(
- () ->
- SessionUtils.doWithoutCommit(
- PolicyVersionMapper.class,
- mapper ->
mapper.insertPolicyVersion(newPolicyPO.getPolicyVersionPO())),
- () ->
- SessionUtils.doWithoutCommit(
- PolicyMetaMapper.class,
- mapper -> mapper.updatePolicyMeta(newPolicyPO,
oldPolicyPO)));
- // we set the updateResult to 1 to indicate that the update is
successful
- updateResult = 1;
- } else {
- updateResult =
- SessionUtils.doWithCommitAndFetchResult(
- PolicyMetaMapper.class,
- mapper -> mapper.updatePolicyMeta(newPolicyPO, oldPolicyPO));
- }
+ POConverters.updatePolicyPOWithVersion(oldPolicyPO,
updatedPolicyEntity);
+ SessionUtils.doMultipleWithCommit(
+ () -> updatePolicyRootWithVersion(ident, oldPolicyPO, newPolicyPO),
+ () ->
+ SessionUtils.doWithoutCommit(
+ PolicyVersionMapper.class,
+ mapper ->
mapper.insertPolicyVersion(newPolicyPO.getPolicyVersionPO())));
} catch (RuntimeException re) {
ExceptionUtils.checkSQLException(
re, Entity.EntityType.POLICY,
updatedPolicyEntity.nameIdentifier().toString());
throw re;
}
- if (updateResult > 0) {
- return updatedPolicyEntity;
- } else {
- throw new IOException("Failed to update the entity: " +
updatedPolicyEntity);
- }
+ return updatedPolicyEntity;
}
@Monitored(
metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
baseMetricName = "deletePolicy")
public boolean deletePolicy(NameIdentifier ident) {
- String metalakeName = ident.namespace().level(0);
- int[] policyMetaDeletedCount = new int[] {0};
- int[] policyVersionDeletedCount = new int[] {0};
+ PolicyPO policyPO;
+ try {
+ policyPO = getPolicyPOByMetalakeAndName(ident.namespace().level(0),
ident.name());
+ } catch (NoSuchEntityException e) {
+ return false;
+ }
+ return deletePolicy(ident, policyPO);
+ }
- // We should delete meta and version info
- SessionUtils.doMultipleWithCommit(
- () ->
- policyMetaDeletedCount[0] =
- SessionUtils.getWithoutCommit(
- PolicyMetaMapper.class,
- mapper ->
-
mapper.softDeletePolicyByMetalakeAndPolicyName(metalakeName, ident.name())),
- () ->
- policyVersionDeletedCount[0] =
- SessionUtils.getWithoutCommit(
- PolicyVersionMapper.class,
- mapper ->
- mapper.softDeletePolicyVersionByMetalakeAndPolicyName(
- metalakeName, ident.name())));
- return policyMetaDeletedCount[0] + policyVersionDeletedCount[0] > 0;
+ /**
+ * Deletes the policy the caller observed. The delete is a compare-and-set
on the observed
+ * version, so a policy that changed since it was read is rejected instead
of being removed, and
+ * the version snapshots and dependent relations are cleaned up in the same
transaction as the
+ * policy row itself. If another transaction deletes or renames the observed
policy 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 #deletePolicy(NameIdentifier)}, which reads it first.
+ */
+ @VisibleForTesting
+ boolean deletePolicy(NameIdentifier ident, PolicyPO policyPO) {
+ long policyId = policyPO.getPolicyId();
+
+ try {
+ SessionUtils.doMultipleWithCommit(
+ () -> deletePolicyWithVersion(ident, policyPO),
+ () ->
+ SessionUtils.doWithoutCommit(
+ PolicyVersionMapper.class,
+ mapper ->
mapper.softDeletePolicyVersionsByPolicyId(policyId)),
+ () ->
+ SessionUtils.doWithoutCommit(
+ PolicyMetadataObjectRelMapper.class,
+ mapper ->
mapper.softDeletePolicyMetadataObjectRelsByPolicyId(policyId)),
+ () ->
+ SessionUtils.doWithoutCommit(
+ PolicyTagRelMapper.class, mapper ->
mapper.softDeleteByPolicyId(policyId)),
+ () ->
+ SessionUtils.doWithoutCommit(
+ TagMetadataObjectRelMapper.class,
+ mapper ->
+ mapper.softDeleteTagMetadataObjectRelsByMetadataObject(
+ policyId, MetadataObject.Type.POLICY.name())),
+ () ->
+ SessionUtils.doWithoutCommit(
+ OwnerMetaMapper.class,
+ mapper ->
+ mapper.softDeleteOwnerRelByMetadataObjectIdAndType(
+ policyId, MetadataObject.Type.POLICY.name())),
+ () ->
+ SessionUtils.doWithoutCommit(
+ SecurableObjectMapper.class,
+ mapper ->
+ mapper.softDeleteObjectRelsByMetadataObject(
+ policyId, MetadataObject.Type.POLICY.name())));
+ return true;
+ } catch (NoSuchEntityException e) {
+ return false;
+ }
}
@Monitored(
@@ -313,75 +340,16 @@ public class PolicyMetaService {
NameIdentifier[] policiesToAdd,
NameIdentifier[] policiesToRemove)
throws NoSuchEntityException, EntityAlreadyExistsException, IOException {
- MetadataObject metadataObject =
NameIdentifierUtil.toMetadataObject(objectIdent, objectType);
- String metalake = objectIdent.namespace().level(0);
-
try {
- Long metadataObjectId = EntityIdService.getEntityId(objectIdent,
objectType);
-
- // Fetch all the policies need to associate with the metadata object.
- List<String> policyNamesToAdd =
-
Arrays.stream(policiesToAdd).map(NameIdentifier::name).collect(Collectors.toList());
- List<PolicyPO> policyPOsToAdd =
- policyNamesToAdd.isEmpty()
- ? Collections.emptyList()
- : getPolicyPOsByMetalakeAndNames(metalake, policyNamesToAdd);
-
- // Fetch all the policies need to remove from the metadata object.
- List<String> policyNamesToRemove =
-
Arrays.stream(policiesToRemove).map(NameIdentifier::name).collect(Collectors.toList());
- List<PolicyPO> policyPOsToRemove =
- policyNamesToRemove.isEmpty()
- ? Collections.emptyList()
- : getPolicyPOsByMetalakeAndNames(metalake, policyNamesToRemove);
-
- SessionUtils.doMultipleWithCommit(
- () -> {
- // Insert the policy metadata object relations.
- if (policyPOsToAdd.isEmpty()) {
- return;
- }
-
- List<PolicyMetadataObjectRelPO> policyRelsToAdd =
- policyPOsToAdd.stream()
- .map(
- policyPO ->
-
POConverters.initializePolicyMetadataObjectRelPOWithVersion(
- policyPO.getPolicyId(),
- metadataObjectId,
- metadataObject.type().toString()))
- .collect(Collectors.toList());
- SessionUtils.doWithoutCommit(
- PolicyMetadataObjectRelMapper.class,
- mapper ->
mapper.batchInsertPolicyMetadataObjectRels(policyRelsToAdd));
- },
- () -> {
- // Remove the policy metadata object relations.
- if (policyPOsToRemove.isEmpty()) {
- return;
- }
-
- List<Long> policyIdsToRemove =
-
policyPOsToRemove.stream().map(PolicyPO::getPolicyId).collect(Collectors.toList());
- SessionUtils.doWithoutCommit(
- PolicyMetadataObjectRelMapper.class,
- mapper ->
-
mapper.batchDeletePolicyMetadataObjectRelsByPolicyIdsAndMetadataObject(
- metadataObjectId, metadataObject.type().toString(),
policyIdsToRemove));
- });
-
- // Fetch all the policies associated with the metadata object after the
operation.
- List<PolicyPO> policyPOs =
- SessionUtils.getWithoutCommit(
- PolicyMetadataObjectRelMapper.class,
- mapper ->
- mapper.listPolicyPOsByMetadataObjectIdAndType(
- metadataObjectId, metadataObject.type().toString()));
-
- return policyPOs.stream()
- .map(policyPO -> POConverters.fromPolicyPO(policyPO,
NamespaceUtil.ofPolicy(metalake)))
- .collect(Collectors.toList());
-
+ // One transaction for the whole association change: the policy rows
stay locked from the
+ // moment they are read until the relation rows are rewritten and read
back, so a conflict
+ // rolls the whole change back instead of leaving a half-applied
association set behind. The
+ // mapper handed to the callback is unused; the call only opens and
closes the transaction.
+ return SessionUtils.doWithCommitAndFetchResult(
+ PolicyMetaMapper.class,
+ ignored ->
+ associatePoliciesWithMetadataObjectWithoutCommit(
+ objectIdent, objectType, policiesToAdd, policiesToRemove));
} catch (RuntimeException e) {
ExceptionUtils.checkSQLException(e, Entity.EntityType.POLICY,
objectIdent.toString());
throw e;
@@ -440,28 +408,6 @@ public class PolicyMetaService {
return totalDeletedCount;
}
- private PolicyPO getPolicyPOByMetalakeAndName(String metalakeName, String
policyName) {
- PolicyPO policyPO =
- SessionUtils.getWithoutCommit(
- PolicyMetaMapper.class,
- mapper -> mapper.selectPolicyMetaByMetalakeAndName(metalakeName,
policyName));
-
- if (policyPO == null) {
- throw new NoSuchEntityException(
- NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
- Entity.EntityType.POLICY.name().toLowerCase(),
- policyName);
- }
- return policyPO;
- }
-
- private List<PolicyPO> getPolicyPOsByMetalakeAndNames(
- String metalakeName, List<String> policyNames) {
- return SessionUtils.getWithoutCommit(
- PolicyMetaMapper.class,
- mapper -> mapper.listPolicyPOsByMetalakeAndPolicyNames(metalakeName,
policyNames));
- }
-
/**
* Get policy id by policy name
*
@@ -500,4 +446,319 @@ public class PolicyMetaService {
return POConverters.fromPolicyPOs(policyPOs, firstIdent.namespace());
});
}
+
+ private List<PolicyEntity> associatePoliciesWithMetadataObjectWithoutCommit(
+ NameIdentifier objectIdent,
+ Entity.EntityType objectType,
+ NameIdentifier[] policiesToAdd,
+ NameIdentifier[] policiesToRemove) {
+ MetadataObject metadataObject =
NameIdentifierUtil.toMetadataObject(objectIdent, objectType);
+ String metalake = objectIdent.namespace().level(0);
+
+ Long metadataObjectId = EntityIdService.getEntityId(objectIdent,
objectType);
+
+ // Fetch all the policies need to associate with the metadata object.
+ List<String> policyNamesToAdd =
+
Arrays.stream(policiesToAdd).map(NameIdentifier::name).collect(Collectors.toList());
+ List<PolicyPO> policyPOsToAdd =
+ policyNamesToAdd.isEmpty()
+ ? Collections.emptyList()
+ : getPolicyPOsByMetalakeAndNames(metalake, policyNamesToAdd);
+
+ // Fetch all the policies need to remove from the metadata object.
+ List<String> policyNamesToRemove =
+
Arrays.stream(policiesToRemove).map(NameIdentifier::name).collect(Collectors.toList());
+ List<PolicyPO> policyPOsToRemove =
+ policyNamesToRemove.isEmpty()
+ ? Collections.emptyList()
+ : getPolicyPOsByMetalakeAndNames(metalake, policyNamesToRemove);
+ Map<Long, PolicyPO> lockedPolicyPOs =
+ lockPoliciesForAssociation(policyPOsToAdd, policyPOsToRemove);
+ policyPOsToAdd = currentPolicyPOs(policyPOsToAdd, lockedPolicyPOs);
+ policyPOsToRemove = currentPolicyPOs(policyPOsToRemove, lockedPolicyPOs);
+
+ if (!policyPOsToAdd.isEmpty()) {
+ List<PolicyMetadataObjectRelPO> policyRelsToAdd =
+ policyPOsToAdd.stream()
+ .map(
+ policyPO ->
+
POConverters.initializePolicyMetadataObjectRelPOWithVersion(
+ policyPO.getPolicyId(),
+ metadataObjectId,
+ metadataObject.type().toString()))
+ .collect(Collectors.toList());
+ SessionUtils.doWithoutCommit(
+ PolicyMetadataObjectRelMapper.class,
+ mapper ->
mapper.batchInsertPolicyMetadataObjectRels(policyRelsToAdd));
+ }
+ if (!policyPOsToRemove.isEmpty()) {
+ List<Long> policyIdsToRemove =
+
policyPOsToRemove.stream().map(PolicyPO::getPolicyId).collect(Collectors.toList());
+ SessionUtils.doWithoutCommit(
+ PolicyMetadataObjectRelMapper.class,
+ mapper ->
+
mapper.batchDeletePolicyMetadataObjectRelsByPolicyIdsAndMetadataObject(
+ metadataObjectId, metadataObject.type().toString(),
policyIdsToRemove));
+ }
+
+ // Fetch all the policies associated with the metadata object after the
operation.
+ List<PolicyPO> policyPOs =
+ SessionUtils.getWithoutCommit(
+ PolicyMetadataObjectRelMapper.class,
+ mapper ->
+ mapper.listPolicyPOsByMetadataObjectIdAndType(
+ metadataObjectId, metadataObject.type().toString()));
+
+ return policyPOs.stream()
+ .map(policyPO -> POConverters.fromPolicyPO(policyPO,
NamespaceUtil.ofPolicy(metalake)))
+ .collect(Collectors.toList());
+ }
+
+ /**
+ * Holds the parent metalake row for the rest of the transaction, so a
policy cannot be created
+ * under a metalake that is going away.
+ *
+ * <p>The lock is shared, not exclusive: many policies 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 lockMetalakeForPolicyCreate(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 the policy row and its content snapshot.
+ *
+ * <p>An overwrite is not an upsert any more: the existing row is located
and locked first, and
+ * the replacement is written as the next version of that row, so the
snapshot history survives
+ * the overwrite instead of being reset. When no row is there to replace,
the overwrite inserts
+ * like a plain create. If another create wins the unique key after the
locking lookup misses, the
+ * caller receives a retryable optimistic-lock failure after the transaction
is rolled back.
+ *
+ * <p>An overwrite no longer revives a soft-deleted row that happens to
carry the same policy ID.
+ * Such a row keeps the primary key, so the insert is rejected as an
already-existing policy,
+ * which is the honest answer: the snapshots and relations of the deleted
policy are gone with it.
+ */
+ private void insertPolicyWithoutCommit(
+ PolicyEntity policyEntity, PolicyPO initializedPolicyPO, boolean
overwritten) {
+ if (!overwritten) {
+ insertNewPolicyWithoutCommit(initializedPolicyPO);
+ return;
+ }
+
+ PolicyPO existingPolicyPO =
findAndLockPolicyForOverwrite(initializedPolicyPO);
+ if (existingPolicyPO == null) {
+ if (SessionUtils.getWithoutCommit(
+ PolicyMetaMapper.class,
+ mapper ->
mapper.countDeletedPolicyMetasById(initializedPolicyPO.getPolicyId()))
+ > 0) {
+ throw new EntityAlreadyExistsException(
+ "The policy ID %s is reserved by a deleted policy; use a new ID",
+ initializedPolicyPO.getPolicyId());
+ }
+ insertNewPolicyWithoutCommit(initializedPolicyPO);
+ return;
+ }
+
+ PolicyPO replacementPolicyPO =
+ POConverters.updatePolicyPOWithVersion(existingPolicyPO,
initializedPolicyPO);
+ NameIdentifier observedIdentifier =
+ NameIdentifier.of(policyEntity.namespace(),
existingPolicyPO.getPolicyName());
+ updatePolicyRootWithVersion(observedIdentifier, existingPolicyPO,
replacementPolicyPO);
+ SessionUtils.doWithoutCommit(
+ PolicyVersionMapper.class,
+ mapper ->
mapper.insertPolicyVersion(replacementPolicyPO.getPolicyVersionPO()));
+ }
+
+ private void insertNewPolicyWithoutCommit(PolicyPO policyPO) {
+ SessionUtils.doWithoutCommit(
+ PolicyMetaMapper.class, mapper -> mapper.insertPolicyMeta(policyPO));
+ SessionUtils.doWithoutCommit(
+ PolicyVersionMapper.class,
+ mapper -> mapper.insertPolicyVersion(policyPO.getPolicyVersionPO()));
+ }
+
+ /**
+ * Resolves and exclusively locks the row an overwrite replaces, or returns
null when the name and
+ * the ID are both free.
+ *
+ * <p>The name is the primary key of the search: an overwrite claims the row
that currently holds
+ * the target name, and the caller-supplied policy ID is dropped in that
case so the row keeps the
+ * stable ID its version snapshots and relation rows point at. A caller that
supplies the ID of
+ * one policy together with the name of another therefore replaces the
content of the policy that
+ * holds the name, not the one the ID identifies. This matches {@code
+ * TagMetaService.findAndLockTagForOverwrite} and is only reachable through
tests today: {@link
+ * org.apache.gravitino.policy.PolicyManager} always writes with {@code
overwritten = false}.
+ *
+ * <p>The lookup by ID is the fallback for a rename-by-overwrite, where the
new name is free and
+ * the row is found by its stable ID.
+ */
+ private PolicyPO findAndLockPolicyForOverwrite(PolicyPO initializedPolicyPO)
{
+ PolicyPO sameNamePolicyPO =
+ SessionUtils.getWithoutCommit(
+ PolicyMetaMapper.class,
+ mapper ->
+ mapper.selectPolicyMetaByMetalakeIdAndNameForUpdate(
+ initializedPolicyPO.getMetalakeId(),
initializedPolicyPO.getPolicyName()));
+ if (sameNamePolicyPO != null) {
+ return sameNamePolicyPO;
+ }
+
+ PolicyPO sameIdPolicyPO =
+ SessionUtils.getWithoutCommit(
+ PolicyMetaMapper.class,
+ mapper ->
mapper.selectPolicyByPolicyIdForUpdate(initializedPolicyPO.getPolicyId()));
+ if (sameIdPolicyPO != null
+ && !Objects.equals(sameIdPolicyPO.getMetalakeId(),
initializedPolicyPO.getMetalakeId())) {
+ throw new EntityAlreadyExistsException(
+ "The policy ID %s already belongs to a different metalake",
+ initializedPolicyPO.getPolicyId());
+ }
+ return sameIdPolicyPO;
+ }
+
+ /**
+ * Advances the policy row to the next version, keyed on the version the
caller observed. A row
+ * that moved on, was renamed away, or was deleted in between matches
nothing, and the failure is
+ * classified as a conflict or as a missing policy by {@link
#policyWriteFailure}.
+ */
+ private void updatePolicyRootWithVersion(
+ NameIdentifier identifier, PolicyPO oldPolicyPO, PolicyPO newPolicyPO) {
+ OccWriteSupport.updateWithVersion(
+ () -> {
+ Integer updated =
+ SessionUtils.getWithoutCommit(
+ PolicyMetaMapper.class,
+ mapper -> mapper.updatePolicyMeta(newPolicyPO, oldPolicyPO));
+ return updated == null ? 0 : updated;
+ },
+ () -> policyWriteFailure(identifier, oldPolicyPO));
+ }
+
+ private void deletePolicyWithVersion(NameIdentifier identifier, PolicyPO
observedPolicyPO) {
+ OccWriteSupport.deleteWithVersion(
+ () ->
+ SessionUtils.getWithoutCommit(
+ PolicyMetaMapper.class,
+ mapper ->
+ mapper.softDeletePolicyByIdAndVersion(
+ observedPolicyPO.getPolicyId(),
observedPolicyPO.getCurrentVersion())),
+ () -> policyWriteFailure(identifier, observedPolicyPO));
+ }
+
+ private RuntimeException policyWriteFailure(
+ NameIdentifier identifier, PolicyPO observedPolicyPO) {
+ return OccWriteSupport.writeFailure(
+ identifier,
+ Entity.EntityType.POLICY,
+ () ->
+ SessionUtils.getWithoutCommit(
+ PolicyMetaMapper.class,
+ mapper ->
mapper.selectPolicyByPolicyIdForUpdate(observedPolicyPO.getPolicyId())),
+ null,
+ current ->
+ Objects.equals(current.getPolicyName(),
observedPolicyPO.getPolicyName())
+ && Objects.equals(current.getMetalakeId(),
observedPolicyPO.getMetalakeId()));
+ }
+
+ /**
+ * Locks every policy taking part in an association change and returns the
rows as they are now,
+ * keyed by policy ID, so the association cannot be written against a policy
that is being renamed
+ * or dropped.
+ *
+ * <p>The rows are locked in policy-ID order. Two association changes that
touch the same policies
+ * therefore take the locks in the same order and queue up instead of
deadlocking.
+ */
+ private Map<Long, PolicyPO> lockPoliciesForAssociation(
+ List<PolicyPO> policyPOsToAdd, List<PolicyPO> policyPOsToRemove) {
+ List<PolicyPO> observedPolicyPOs = new ArrayList<>(policyPOsToAdd);
+ observedPolicyPOs.addAll(policyPOsToRemove);
+ return lockPolicies(observedPolicyPOs);
+ }
+
+ /**
+ * Locks the given policy rows and returns them as they are now, keyed by
policy ID.
+ *
+ * <p>The rows are locked by one statement that orders them by policy ID, so
callers that touch
+ * overlapping policies take the row locks in the same order and queue up
instead of deadlocking,
+ * and a change touching many policies 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 policy name, and
that name no longer
+ * points at this row.
+ */
+ static Map<Long, PolicyPO> lockPolicies(List<PolicyPO> observedPolicyPOs) {
+ Map<Long, PolicyPO> observedById = new LinkedHashMap<>();
+ observedPolicyPOs.forEach(policyPO ->
observedById.put(policyPO.getPolicyId(), policyPO));
+ if (observedById.isEmpty()) {
+ return new LinkedHashMap<>();
+ }
+
+ List<PolicyPO> lockedRows =
+ SessionUtils.getWithoutCommit(
+ PolicyMetaMapper.class,
+ mapper ->
+ mapper.listPolicyPOsByPolicyIdsForUpdate(new
ArrayList<>(observedById.keySet())));
+ Map<Long, PolicyPO> lockedById = new LinkedHashMap<>();
+ lockedRows.forEach(policyPO -> lockedById.put(policyPO.getPolicyId(),
policyPO));
+
+ Map<Long, PolicyPO> lockedPolicyPOs = new LinkedHashMap<>();
+ for (PolicyPO observedPolicyPO : observedById.values()) {
+ PolicyPO lockedPolicyPO =
+ OccWriteSupport.lockParentForChildWrite(
+ observedPolicyPO.getPolicyName(),
+ Entity.EntityType.POLICY,
+ () -> lockedById.get(observedPolicyPO.getPolicyId()),
+ null,
+ current ->
+ Objects.equals(current.getPolicyName(),
observedPolicyPO.getPolicyName())
+ && Objects.equals(current.getMetalakeId(),
observedPolicyPO.getMetalakeId()));
+ lockedPolicyPOs.put(lockedPolicyPO.getPolicyId(), lockedPolicyPO);
+ }
+ return lockedPolicyPOs;
+ }
+
+ private static List<PolicyPO> currentPolicyPOs(
+ List<PolicyPO> observedPolicyPOs, Map<Long, PolicyPO> lockedPolicyPOs) {
+ return observedPolicyPOs.stream()
+ .map(policyPO -> lockedPolicyPOs.get(policyPO.getPolicyId()))
+ .collect(Collectors.toList());
+ }
+
+ private PolicyPO getPolicyPOByMetalakeAndName(String metalakeName, String
policyName) {
+ PolicyPO policyPO =
+ SessionUtils.getWithoutCommit(
+ PolicyMetaMapper.class,
+ mapper -> mapper.selectPolicyMetaByMetalakeAndName(metalakeName,
policyName));
+
+ if (policyPO == null) {
+ throw new NoSuchEntityException(
+ NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
+ Entity.EntityType.POLICY.name().toLowerCase(),
+ policyName);
+ }
+ return policyPO;
+ }
+
+ private List<PolicyPO> getPolicyPOsByMetalakeAndNames(
+ String metalakeName, List<String> policyNames) {
+ return SessionUtils.getWithoutCommit(
+ PolicyMetaMapper.class,
+ mapper -> mapper.listPolicyPOsByMetalakeAndPolicyNames(metalakeName,
policyNames));
+ }
}
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 49ae66a8d9..886fd565e6 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
@@ -25,6 +25,7 @@ import java.time.Instant;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
@@ -359,7 +360,9 @@ public class PolicyTagRelService {
.map(target -> target.nameIdentifier().name())
.forEach(policyNames::add);
if (policyNames.isEmpty()) {
- return Collections.emptyMap();
+ // A mutable map, like the one built below: returning
Collections.emptyMap() here would mix
+ // mutable and immutable return values, which Error Prone rejects.
+ return new LinkedHashMap<>();
}
List<PolicyPO> policies =
@@ -368,8 +371,13 @@ public class PolicyTagRelService {
mapper ->
mapper.listPolicyPOsByMetalakeAndPolicyNames(
metalake, new ArrayList<>(policyNames)));
- Map<String, Long> policyIds =
- policies.stream().collect(Collectors.toMap(PolicyPO::getPolicyName,
PolicyPO::getPolicyId));
+ // Lock the policy rows in policy-ID order so that two relation changes
touching the same
+ // policies queue up instead of deadlocking. The tag row is locked before
this, so every path
+ // through this service takes its locks in the same tag-then-policy order.
+ Map<String, Long> policyIds = new LinkedHashMap<>();
+ PolicyMetaService.lockPolicies(policies)
+ .values()
+ .forEach(policy -> policyIds.put(policy.getPolicyName(),
policy.getPolicyId()));
for (String policyName : policyNames) {
if (!policyIds.containsKey(policyName)) {
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 07f98776c8..9f87f4c0fb 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
@@ -64,7 +64,6 @@ 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.rel.Column;
import org.apache.gravitino.rel.Table;
import org.apache.gravitino.rel.expressions.Expression;
@@ -761,64 +760,56 @@ public class POConverters {
}
}
- public static boolean checkPolicyVersionNeedUpdate(
- PolicyVersionPO oldPolicyVersionPO, PolicyEntity newPolicy) {
- if (!StringUtils.equals(oldPolicyVersionPO.getPolicyComment(),
newPolicy.comment())
- || oldPolicyVersionPO.isEnabled() != newPolicy.enabled()) {
- return true;
- }
-
+ /**
+ * Builds the next complete policy metadata and content snapshot.
+ *
+ * <p>The row keeps the ID it already has: {@code oldPolicyPO} is the row
being replaced, and its
+ * ID is what the version snapshots and every relation row point at. An
alter cannot change the
+ * ID, because {@code PolicyMetaService.updatePolicy} rejects an updater
that returns a different
+ * one; an overwrite of a name held by another row deliberately updates that
row rather than
+ * inserting a second one under the same name, so the ID the caller supplied
is dropped.
+ *
+ * @param oldPolicyPO The policy row observed by the caller.
+ * @param newPolicy The policy values to persist.
+ * @return The policy row and version snapshot at the next monotonic version.
+ */
+ public static PolicyPO updatePolicyPOWithVersion(PolicyPO oldPolicyPO,
PolicyEntity newPolicy) {
try {
- PolicyContent oldContent =
- JsonUtils.anyFieldMapper()
- .readValue(oldPolicyVersionPO.getContent(),
newPolicy.policyType().contentClass());
- if (oldContent == null) {
- return newPolicy.content() != null;
- }
- return !oldContent.equals(newPolicy.content());
+ return buildNextPolicyPOVersion(
+ oldPolicyPO,
+ newPolicy.name(),
+ newPolicy.policyType().policyType(),
+ JsonUtils.anyFieldMapper().writeValueAsString(newPolicy.auditInfo()),
+ newPolicy.comment(),
+ newPolicy.enabled(),
+ JsonUtils.anyFieldMapper().writeValueAsString(newPolicy.content()));
} catch (JsonProcessingException e) {
- throw new RuntimeException("Failed to deserialize json object:", e);
+ throw new RuntimeException("Failed to serialize json object:", e);
}
}
+ /**
+ * Builds the next policy version from values that were serialized before
acquiring a row lock.
+ *
+ * <p>This overload is used by overwrite: the initialized replacement
already contains the
+ * serialized audit and content values, so advancing the locked row does not
repeat CPU-bound JSON
+ * serialization while other writers wait for the lock.
+ *
+ * @param oldPolicyPO The locked policy row being replaced.
+ * @param replacementPolicyPO The initialized replacement values.
+ * @return The policy row and version snapshot at the next monotonic version.
+ */
public static PolicyPO updatePolicyPOWithVersion(
- PolicyPO oldPolicyPO, PolicyEntity newPolicy, boolean needUpdateVersion)
{
- try {
- Long lastVersion = oldPolicyPO.getLastVersion();
- Long currentVersion;
- PolicyVersionPO newPolicyVersionPO;
- // Will set the version to the last version + 1
- if (needUpdateVersion) {
- lastVersion++;
- currentVersion = lastVersion;
- newPolicyVersionPO =
- PolicyVersionPO.builder()
- .withMetalakeId(oldPolicyPO.getMetalakeId())
- .withPolicyId(newPolicy.id())
- .withVersion(currentVersion)
- .withPolicyComment(newPolicy.comment())
- .withEnabled(newPolicy.enabled())
-
.withContent(JsonUtils.anyFieldMapper().writeValueAsString(newPolicy.content()))
- .withDeletedAt(DEFAULT_DELETED_AT)
- .build();
- } else {
- currentVersion = oldPolicyPO.getCurrentVersion();
- newPolicyVersionPO = oldPolicyPO.getPolicyVersionPO();
- }
- return PolicyPO.builder()
- .withPolicyId(newPolicy.id())
- .withPolicyName(newPolicy.name())
- .withPolicyType(newPolicy.policyType().policyType())
- .withMetalakeId(oldPolicyPO.getMetalakeId())
-
.withAuditInfo(JsonUtils.anyFieldMapper().writeValueAsString(newPolicy.auditInfo()))
- .withCurrentVersion(currentVersion)
- .withLastVersion(lastVersion)
- .withDeletedAt(DEFAULT_DELETED_AT)
- .withPolicyVersionPO(newPolicyVersionPO)
- .build();
- } catch (JsonProcessingException e) {
- throw new RuntimeException("Failed to serialize json object:", e);
- }
+ PolicyPO oldPolicyPO, PolicyPO replacementPolicyPO) {
+ PolicyVersionPO replacementVersionPO =
replacementPolicyPO.getPolicyVersionPO();
+ return buildNextPolicyPOVersion(
+ oldPolicyPO,
+ replacementPolicyPO.getPolicyName(),
+ replacementPolicyPO.getPolicyType(),
+ replacementPolicyPO.getAuditInfo(),
+ replacementVersionPO.getPolicyComment(),
+ replacementVersionPO.isEnabled(),
+ replacementVersionPO.getContent());
}
/**
@@ -1819,6 +1810,38 @@ public class POConverters {
.collect(Collectors.toList());
}
+ private static PolicyPO buildNextPolicyPOVersion(
+ PolicyPO oldPolicyPO,
+ String policyName,
+ String policyType,
+ String auditInfo,
+ String policyComment,
+ boolean enabled,
+ String content) {
+ Long nextVersion = Math.max(oldPolicyPO.getCurrentVersion(),
oldPolicyPO.getLastVersion()) + 1;
+ PolicyVersionPO newPolicyVersionPO =
+ PolicyVersionPO.builder()
+ .withMetalakeId(oldPolicyPO.getMetalakeId())
+ .withPolicyId(oldPolicyPO.getPolicyId())
+ .withVersion(nextVersion)
+ .withPolicyComment(policyComment)
+ .withEnabled(enabled)
+ .withContent(content)
+ .withDeletedAt(DEFAULT_DELETED_AT)
+ .build();
+ return PolicyPO.builder()
+ .withPolicyId(oldPolicyPO.getPolicyId())
+ .withPolicyName(policyName)
+ .withPolicyType(policyType)
+ .withMetalakeId(oldPolicyPO.getMetalakeId())
+ .withAuditInfo(auditInfo)
+ .withCurrentVersion(nextVersion)
+ .withLastVersion(nextVersion)
+ .withDeletedAt(DEFAULT_DELETED_AT)
+ .withPolicyVersionPO(newPolicyVersionPO)
+ .build();
+ }
+
private static ModelVersionAliasRelPO createAliasRelPO(Long modelId, int
version, String alias) {
return ModelVersionAliasRelPO.builder()
.withModelVersion(version)
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestOccWriteSupport.java
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestOccWriteSupport.java
index e4432034b8..8e1ecbf01b 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestOccWriteSupport.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestOccWriteSupport.java
@@ -119,6 +119,23 @@ public class TestOccWriteSupport {
NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
"metalake", "test")));
}
+ @Test
+ void testUpdateWithVersionSuccess() {
+ assertDoesNotThrow(
+ () ->
+ OccWriteSupport.updateWithVersion(
+ () -> 1, () -> new RuntimeException("Should not be thrown")));
+ }
+
+ @Test
+ void testUpdateWithVersionThrowsOnMiss() {
+ assertThrows(
+ OptimisticLockException.class,
+ () ->
+ OccWriteSupport.updateWithVersion(
+ () -> 0, () -> new OptimisticLockException("test conflict")));
+ }
+
@Test
void testDeleteChildrenWithVersionsEmptyOrNull() {
NameIdentifier parentIdent = NameIdentifier.of("parent");
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestPolicyMetaService.java
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestPolicyMetaService.java
index 5dd6056e41..ec4d530da3 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestPolicyMetaService.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestPolicyMetaService.java
@@ -25,6 +25,7 @@ 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;
import java.sql.ResultSet;
@@ -34,28 +35,50 @@ import java.time.Instant;
import java.util.List;
import java.util.Map;
import java.util.Set;
+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 java.util.stream.Collectors;
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.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.AuditInfo;
import org.apache.gravitino.meta.BaseMetalake;
import org.apache.gravitino.meta.CatalogEntity;
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.Policy;
import org.apache.gravitino.policy.PolicyContent;
import org.apache.gravitino.policy.PolicyContents;
import org.apache.gravitino.storage.RandomIdGenerator;
import org.apache.gravitino.storage.relational.TestJDBCBackend;
+import org.apache.gravitino.storage.relational.mapper.PolicyMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.PolicyVersionMapper;
+import org.apache.gravitino.storage.relational.po.PolicyPO;
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.utils.NameIdentifierUtil;
import org.apache.gravitino.utils.NamespaceUtil;
import org.apache.ibatis.session.SqlSession;
@@ -417,6 +440,484 @@ public class TestPolicyMetaService extends
TestJDBCBackend {
assertEquals(policyEntity2, loadedPolicyEntity1);
}
+ @TestTemplate
+ public void testMetadataOnlyPolicyAlterCreatesCompleteSnapshot() throws
IOException {
+ createAndInsertMakeLake(METALAKE_NAME);
+ PolicyMetaService policyMetaService = PolicyMetaService.getInstance();
+ PolicyEntity policy =
+ createPolicy(
+ RandomIdGenerator.INSTANCE.nextId(),
+ NamespaceUtil.ofPolicy(METALAKE_NAME),
+ "policy_metadata_occ",
+ AUDIT_INFO);
+ policyMetaService.insertPolicy(policy, false);
+ PolicyPO initialPO = getPolicyPO(policy.nameIdentifier());
+
+ AuditInfo updatedAudit =
+
AuditInfo.builder().withCreator("updated-creator").withCreateTime(Instant.now()).build();
+ PolicyEntity metadataOnlyUpdate =
+ copyPolicy(policy, policy.name(), policy.comment(), updatedAudit);
+ policyMetaService.updatePolicy(policy.nameIdentifier(), ignored ->
metadataOnlyUpdate);
+
+ PolicyPO updatedPO = getPolicyPO(policy.nameIdentifier());
+ assertEquals(initialPO.getCurrentVersion() + 1,
updatedPO.getCurrentVersion().longValue());
+ assertEquals(updatedPO.getCurrentVersion(), updatedPO.getLastVersion());
+ assertEquals(updatedPO.getCurrentVersion(),
updatedPO.getPolicyVersionPO().getVersion());
+ assertEquals(policy.comment(),
updatedPO.getPolicyVersionPO().getPolicyComment());
+ assertEquals(policy.enabled(), updatedPO.getPolicyVersionPO().isEnabled());
+ assertEquals(
+ initialPO.getPolicyVersionPO().getContent(),
updatedPO.getPolicyVersionPO().getContent());
+ assertEquals(2, listPolicyVersions(policy.id()).size());
+ }
+
+ @TestTemplate
+ public void testPolicyOverwriteAdvancesVersionAndRetainsHistory() throws
IOException {
+ createAndInsertMakeLake(METALAKE_NAME);
+ PolicyMetaService policyMetaService = PolicyMetaService.getInstance();
+ PolicyEntity policy =
+ createPolicy(
+ RandomIdGenerator.INSTANCE.nextId(),
+ NamespaceUtil.ofPolicy(METALAKE_NAME),
+ "policy_overwrite_occ",
+ AUDIT_INFO);
+ policyMetaService.insertPolicy(policy, false);
+ PolicyPO initialPO = getPolicyPO(policy.nameIdentifier());
+
+ PolicyEntity replacement = copyPolicy(policy,
"policy_overwrite_occ_renamed", "replacement");
+ policyMetaService.insertPolicy(replacement, true);
+
+ PolicyPO overwrittenPO = getPolicyPO(replacement.nameIdentifier());
+ assertEquals(initialPO.getCurrentVersion() + 1,
overwrittenPO.getCurrentVersion().longValue());
+ assertEquals(overwrittenPO.getCurrentVersion(),
overwrittenPO.getLastVersion());
+ assertEquals(2, listPolicyVersions(policy.id()).size());
+ assertEquals(
+ replacement,
policyMetaService.getPolicyByIdentifier(replacement.nameIdentifier()));
+ }
+
+ @TestTemplate
+ public void
testPolicyAlterReportsOptimisticLockConflictWithoutOrphanVersion()
+ throws IOException {
+ createAndInsertMakeLake(METALAKE_NAME);
+ PolicyMetaService policyMetaService = PolicyMetaService.getInstance();
+ PolicyEntity policy =
+ createPolicy(
+ RandomIdGenerator.INSTANCE.nextId(),
+ NamespaceUtil.ofPolicy(METALAKE_NAME),
+ "policy_alter_occ",
+ AUDIT_INFO);
+ policyMetaService.insertPolicy(policy, false);
+
+ assertThrows(
+ OptimisticLockException.class,
+ () ->
+ policyMetaService.updatePolicy(
+ policy.nameIdentifier(),
+ entity -> {
+ PolicyEntity current = (PolicyEntity) entity;
+ PolicyPO currentPO = getPolicyPO(current.nameIdentifier());
+ PolicyEntity competing = copyPolicy(current, current.name(),
"competing");
+ PolicyPO competingPO =
+ POConverters.updatePolicyPOWithVersion(currentPO,
competing);
+ SessionUtils.doMultipleWithCommit(
+ () ->
+ assertEquals(
+ Integer.valueOf(1),
+ SessionUtils.getWithoutCommit(
+ PolicyMetaMapper.class,
+ mapper ->
mapper.updatePolicyMeta(competingPO, currentPO))),
+ () ->
+ SessionUtils.doWithoutCommit(
+ PolicyVersionMapper.class,
+ mapper ->
+
mapper.insertPolicyVersion(competingPO.getPolicyVersionPO())));
+ return copyPolicy(current, current.name(), "requested");
+ }));
+
+ assertEquals(2, listPolicyVersions(policy.id()).size());
+ assertEquals(
+ "competing",
policyMetaService.getPolicyByIdentifier(policy.nameIdentifier()).comment());
+ }
+
+ @TestTemplate
+ public void testStalePolicyDeleteRollsBackRelationshipCleanup() throws
IOException {
+ createAndInsertMakeLake(METALAKE_NAME);
+ CatalogEntity catalog = createAndInsertCatalog(METALAKE_NAME,
"catalog_policy_delete_occ");
+ PolicyMetaService policyMetaService = PolicyMetaService.getInstance();
+ PolicyEntity policy =
+ createPolicy(
+ RandomIdGenerator.INSTANCE.nextId(),
+ NamespaceUtil.ofPolicy(METALAKE_NAME),
+ "policy_delete_occ",
+ AUDIT_INFO);
+ policyMetaService.insertPolicy(policy, false);
+ policyMetaService.associatePoliciesWithMetadataObject(
+ catalog.nameIdentifier(),
+ catalog.type(),
+ new NameIdentifier[] {policy.nameIdentifier()},
+ new NameIdentifier[0]);
+ PolicyPO stalePO = getPolicyPO(policy.nameIdentifier());
+ policyMetaService.updatePolicy(
+ policy.nameIdentifier(),
+ entity -> copyPolicy((PolicyEntity) entity, ((PolicyEntity)
entity).name(), "updated"));
+
+ assertThrows(
+ OptimisticLockException.class,
+ () -> policyMetaService.deletePolicy(policy.nameIdentifier(),
stalePO));
+ assertEquals(1, countActivePolicyRel(policy.id()));
+ assertTrue(backend.exists(policy.nameIdentifier(),
Entity.EntityType.POLICY));
+ assertEquals(
+ 2,
+ listPolicyVersions(policy.id()).values().stream().filter(v ->
v.longValue() == 0L).count());
+
+ assertTrue(policyMetaService.deletePolicy(policy.nameIdentifier()));
+ assertEquals(0, countActivePolicyRel(policy.id()));
+ assertEquals(
+ 0,
+ listPolicyVersions(policy.id()).values().stream().filter(v ->
v.longValue() == 0L).count());
+ }
+
+ @TestTemplate
+ public void testPolicyCreateIsFencedByParentMetalake() {
+ PolicyMetaService policyMetaService = PolicyMetaService.getInstance();
+ PolicyEntity policy =
+ createPolicy(
+ RandomIdGenerator.INSTANCE.nextId(),
+ NamespaceUtil.ofPolicy("metalake_that_does_not_exist"),
+ "policy_without_metalake",
+ AUDIT_INFO);
+
+ assertThrows(NoSuchEntityException.class, () ->
policyMetaService.insertPolicy(policy, false));
+ assertThrows(NoSuchEntityException.class, () ->
policyMetaService.insertPolicy(policy, true));
+ }
+
+ @TestTemplate
+ public void testPolicyOverwriteReplacesTheRowHoldingTheName() throws
IOException {
+ createAndInsertMakeLake(METALAKE_NAME);
+ PolicyMetaService policyMetaService = PolicyMetaService.getInstance();
+ PolicyEntity policy =
+ createPolicy(
+ RandomIdGenerator.INSTANCE.nextId(),
+ NamespaceUtil.ofPolicy(METALAKE_NAME),
+ "policy_overwrite_by_name",
+ AUDIT_INFO);
+ policyMetaService.insertPolicy(policy, false);
+ PolicyPO initialPO = getPolicyPO(policy.nameIdentifier());
+
+ // A different ID for a name that is already taken replaces the row that
holds the name,
+ // instead of inserting a second row for it.
+ PolicyEntity sameNameOtherId =
+ createPolicy(
+ RandomIdGenerator.INSTANCE.nextId(),
+ NamespaceUtil.ofPolicy(METALAKE_NAME),
+ "policy_overwrite_by_name",
+ AUDIT_INFO);
+ policyMetaService.insertPolicy(sameNameOtherId, true);
+
+ PolicyPO overwrittenPO = getPolicyPO(policy.nameIdentifier());
+ assertEquals(policy.id(), overwrittenPO.getPolicyId().longValue());
+ assertEquals(initialPO.getCurrentVersion() + 1,
overwrittenPO.getCurrentVersion().longValue());
+ assertEquals(2, listPolicyVersions(policy.id()).size());
+ }
+
+ /** A deleted primary key must not be classified as a retryable overwrite
race. */
+ @TestTemplate
+ public void testOverwriteRejectsDeletedPolicyId() throws IOException {
+ createAndInsertMakeLake(METALAKE_NAME);
+ PolicyMetaService service = PolicyMetaService.getInstance();
+ Namespace ns = NamespaceUtil.ofPolicy(METALAKE_NAME);
+ PolicyEntity policy =
+ createPolicy(RandomIdGenerator.INSTANCE.nextId(), ns,
"deleted_policy_id", AUDIT_INFO);
+ service.insertPolicy(policy, false);
+ assertTrue(service.deletePolicy(policy.nameIdentifier()));
+ EntityAlreadyExistsException failure =
+ assertThrows(EntityAlreadyExistsException.class, () ->
service.insertPolicy(policy, true));
+ assertTrue(failure.getMessage().contains("use a new ID"));
+ assertThrows(
+ NoSuchEntityException.class, () ->
service.getPolicyByIdentifier(policy.nameIdentifier()));
+ listPolicyVersions(policy.id()).values().forEach(deletedAt ->
assertTrue(deletedAt > 0));
+
+ PolicyEntity replacement =
+ createPolicy(RandomIdGenerator.INSTANCE.nextId(), ns, policy.name(),
AUDIT_INFO);
+ service.insertPolicy(replacement, true);
+ assertEquals(replacement.id(),
service.getPolicyByIdentifier(policy.nameIdentifier()).id());
+ }
+
+ /** An overwrite cannot adopt a stable ID from a metalake it has not locked.
*/
+ @TestTemplate
+ public void testOverwriteRejectsPolicyIdInAnotherMetalake() throws
IOException {
+ createAndInsertMakeLake(METALAKE_NAME);
+ createAndInsertMakeLake("foreign_policy_metalake");
+ PolicyMetaService service = PolicyMetaService.getInstance();
+ PolicyEntity foreign =
+ createPolicy(
+ RandomIdGenerator.INSTANCE.nextId(),
+ NamespaceUtil.ofPolicy("foreign_policy_metalake"),
+ "foreign_policy",
+ AUDIT_INFO);
+ service.insertPolicy(foreign, false);
+ PolicyEntity incoming =
+ createPolicy(
+ foreign.id(), NamespaceUtil.ofPolicy(METALAKE_NAME),
foreign.name(), AUDIT_INFO);
+ assertThrows(EntityAlreadyExistsException.class, () ->
service.insertPolicy(incoming, true));
+ assertEquals(foreign.id(),
service.getPolicyByIdentifier(foreign.nameIdentifier()).id());
+ assertThrows(
+ NoSuchEntityException.class,
+ () -> service.getPolicyByIdentifier(incoming.nameIdentifier()));
+ }
+
+ /** Two first-time overwrites must serialize or expose a retryable insert
conflict. */
+ @TestTemplate
+ public void testConcurrentOverwriteOfMissingPolicy() throws Exception {
+ createAndInsertMakeLake(METALAKE_NAME);
+ PolicyMetaService service = PolicyMetaService.getInstance();
+ Namespace ns = NamespaceUtil.ofPolicy(METALAKE_NAME);
+ PolicyEntity first =
+ createPolicy(RandomIdGenerator.INSTANCE.nextId(), ns,
"first_overwrite", AUDIT_INFO);
+ PolicyEntity second =
+ copyPolicy(
+ createPolicy(RandomIdGenerator.INSTANCE.nextId(), ns,
first.name(), AUDIT_INFO),
+ first.name(),
+ "second overwrite");
+ CountDownLatch firstWritten = new CountDownLatch(1);
+ CountDownLatch allowCommit = new CountDownLatch(1);
+ CountDownLatch secondStarted = new CountDownLatch(1);
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ Future<Throwable> firstResult =
+ executor.submit(
+ () -> {
+ SessionUtils.beginTransaction();
+ try {
+ service.insertPolicy(first, true);
+ firstWritten.countDown();
+ await(allowCommit);
+ SessionUtils.commitTransaction();
+ return null;
+ } catch (Throwable failure) {
+ SessionUtils.rollbackTransaction();
+ return failure;
+ }
+ });
+ try {
+ assertTrue(firstWritten.await(30, TimeUnit.SECONDS));
+ Future<Throwable> secondResult =
+ executor.submit(
+ () -> {
+ secondStarted.countDown();
+ try {
+ service.insertPolicy(second, true);
+ return null;
+ } catch (Throwable failure) {
+ return failure;
+ }
+ });
+ assertTrue(secondStarted.await(30, TimeUnit.SECONDS));
+ assertThrows(TimeoutException.class, () -> secondResult.get(500,
TimeUnit.MILLISECONDS));
+ allowCommit.countDown();
+ Assertions.assertNull(firstResult.get(30, TimeUnit.SECONDS));
+ Throwable failure = secondResult.get(30, TimeUnit.SECONDS);
+ if (failure != null) {
+ Assertions.assertInstanceOf(OptimisticLockException.class, failure);
+ assertEquals(1, listPolicyVersions(first.id()).size());
+ assertTrue(listPolicyVersions(second.id()).isEmpty());
+ service.insertPolicy(second, true);
+ }
+ PolicyEntity stored =
service.getPolicyByIdentifier(first.nameIdentifier());
+ assertEquals(first.id(), stored.id());
+ assertEquals("second overwrite", stored.comment());
+ assertEquals(2L,
getPolicyPO(first.nameIdentifier()).getCurrentVersion());
+ assertEquals(2, listPolicyVersions(first.id()).size());
+ assertTrue(listPolicyVersions(second.id()).isEmpty());
+ } finally {
+ allowCommit.countDown();
+ executor.shutdownNow();
+ }
+ }
+
+ @TestTemplate
+ public void testPolicyOverwriteByNameDoesNotRevertConcurrentRename() throws
Exception {
+ createAndInsertMakeLake(METALAKE_NAME);
+ PolicyMetaService policyMetaService = PolicyMetaService.getInstance();
+ PolicyEntity original =
+ createPolicy(
+ RandomIdGenerator.INSTANCE.nextId(),
+ NamespaceUtil.ofPolicy(METALAKE_NAME),
+ "policy_overwrite_rename_race",
+ AUDIT_INFO);
+ policyMetaService.insertPolicy(original, false);
+ PolicyPO observedPO = getPolicyPO(original.nameIdentifier());
+
+ PolicyEntity renamed = copyPolicy(original,
"policy_overwrite_rename_winner", "rename winner");
+ PolicyPO renamedPO = POConverters.updatePolicyPOWithVersion(observedPO,
renamed);
+ PolicyEntity replacement =
+ createPolicy(
+ RandomIdGenerator.INSTANCE.nextId(),
+ NamespaceUtil.ofPolicy(METALAKE_NAME),
+ original.name(),
+ AUDIT_INFO);
+
+ 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(
+ () ->
+ assertEquals(
+ Integer.valueOf(1),
+ SessionUtils.getWithoutCommit(
+ PolicyMetaMapper.class,
+ mapper -> mapper.updatePolicyMeta(renamedPO,
observedPO))),
+ () ->
+ SessionUtils.doWithoutCommit(
+ PolicyVersionMapper.class,
+ mapper ->
mapper.insertPolicyVersion(renamedPO.getPolicyVersionPO())),
+ () -> {
+ 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 {
+ policyMetaService.insertPolicy(replacement, true);
+ return null;
+ } catch (Throwable throwable) {
+ return throwable;
+ }
+ });
+ assertTrue(overwriteStarted.await(30, TimeUnit.SECONDS));
+ 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();
+ }
+
+ assertEquals(
+ original.id(),
policyMetaService.getPolicyByIdentifier(renamed.nameIdentifier()).id());
+ assertEquals(
+ replacement.id(),
policyMetaService.getPolicyByIdentifier(original.nameIdentifier()).id());
+ }
+
+ @TestTemplate
+ public void testPolicyDeleteReturnsFalseWhenConcurrentDeleteWins() throws
IOException {
+ createAndInsertMakeLake(METALAKE_NAME);
+ PolicyMetaService policyMetaService = PolicyMetaService.getInstance();
+ PolicyEntity policy =
+ createPolicy(
+ RandomIdGenerator.INSTANCE.nextId(),
+ NamespaceUtil.ofPolicy(METALAKE_NAME),
+ "policy_concurrent_delete",
+ AUDIT_INFO);
+ policyMetaService.insertPolicy(policy, false);
+ PolicyPO observedPO = getPolicyPO(policy.nameIdentifier());
+
+ assertTrue(policyMetaService.deletePolicy(policy.nameIdentifier()));
+ assertFalse(policyMetaService.deletePolicy(policy.nameIdentifier(),
observedPO));
+ }
+
+ @TestTemplate
+ public void testDeletePolicyCleansEveryDependentRelation() throws
IOException {
+ createAndInsertMakeLake(METALAKE_NAME);
+ CatalogEntity catalog = createAndInsertCatalog(METALAKE_NAME,
"catalog_policy_cascade");
+ PolicyMetaService policyMetaService = PolicyMetaService.getInstance();
+ PolicyEntity policy =
+ createPolicy(
+ RandomIdGenerator.INSTANCE.nextId(),
+ NamespaceUtil.ofPolicy(METALAKE_NAME),
+ "policy_cascade_occ",
+ AUDIT_INFO);
+ policyMetaService.insertPolicy(policy, false);
+ policyMetaService.associatePoliciesWithMetadataObject(
+ catalog.nameIdentifier(),
+ catalog.type(),
+ new NameIdentifier[] {policy.nameIdentifier()},
+ new NameIdentifier[0]);
+
+ TagEntity tag = createAndInsertTagEntity("tag_policy_cascade", "tag
comment", 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]));
+ TagMetaService.getInstance()
+ .associateTagsWithMetadataObject(
+ policy.nameIdentifier(),
+ Entity.EntityType.POLICY,
+ new NameIdentifier[] {tag.nameIdentifier()},
+ new NameIdentifier[0]);
+
+ UserEntity user =
+ createUserEntity(
+ RandomIdGenerator.INSTANCE.nextId(),
+ AuthorizationUtils.ofUserNamespace(METALAKE_NAME),
+ "user_policy_cascade",
+ AUDIT_INFO);
+ backend.insert(user, false);
+ OwnerMetaService.getInstance()
+ .setOwner(
+ policy.nameIdentifier(), Entity.EntityType.POLICY,
user.nameIdentifier(), user.type());
+
+ RoleEntity role =
+ createRoleEntity(
+ RandomIdGenerator.INSTANCE.nextId(),
+ AuthorizationUtils.ofRoleNamespace(METALAKE_NAME),
+ "role_policy_cascade",
+ AUDIT_INFO,
+ Lists.newArrayList(
+ SecurableObjects.ofPolicy(
+ policy.name(),
Lists.newArrayList(Privileges.ApplyPolicy.allow()))),
+ null);
+ backend.insert(role, false);
+
+ String policyAsMetadataObject =
+ String.format("metadata_object_id = %d AND metadata_object_type =
'POLICY'", policy.id());
+ assertEquals(1, countActivePolicyRel(policy.id()));
+ assertEquals(1, countActiveRows("policy_tag_relation_meta", "policy_id = "
+ policy.id()));
+ assertEquals(1, countActiveRows("tag_relation_meta",
policyAsMetadataObject));
+ assertEquals(1, countActiveRows("owner_meta", policyAsMetadataObject));
+ assertEquals(
+ 1,
+ countActiveRows(
+ "role_meta_securable_object",
+ String.format("metadata_object_id = %d AND type = 'POLICY'",
policy.id())));
+
+ assertTrue(policyMetaService.deletePolicy(policy.nameIdentifier()));
+
+ assertEquals(0, countActivePolicyRel(policy.id()));
+ assertEquals(0, countActiveRows("policy_tag_relation_meta", "policy_id = "
+ policy.id()));
+ assertEquals(0, countActiveRows("tag_relation_meta",
policyAsMetadataObject));
+ assertEquals(0, countActiveRows("owner_meta", policyAsMetadataObject));
+ assertEquals(
+ 0,
+ countActiveRows(
+ "role_meta_securable_object",
+ String.format("metadata_object_id = %d AND type = 'POLICY'",
policy.id())));
+ assertEquals(0, listPolicyVersions(policy.id()).values().stream().filter(v
-> v == 0L).count());
+ }
+
@TestTemplate
public void testDeletePolicy() throws IOException {
createAndInsertMakeLake(METALAKE_NAME);
@@ -1040,26 +1541,28 @@ public class TestPolicyMetaService extends
TestJDBCBackend {
return new EntitiesToTest(catalog, schema, table, topic, fileset, model);
}
- private Integer countActivePolicyRel(Long policyId) {
+ private int countActiveRows(String table, String whereClause) {
try (SqlSession sqlSession =
SqlSessionFactoryHelper.getInstance().getSqlSessionFactory().openSession(true);
Connection connection = sqlSession.getConnection();
- Statement statement1 = connection.createStatement();
- ResultSet rs1 =
- statement1.executeQuery(
+ Statement statement = connection.createStatement();
+ ResultSet rs =
+ statement.executeQuery(
String.format(
- "SELECT count(*) FROM policy_relation_meta WHERE policy_id
= %d AND deleted_at = 0",
- policyId))) {
- if (rs1.next()) {
- return rs1.getInt(1);
- } else {
- throw new RuntimeException("Doesn't contain data");
+ "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 countActivePolicyRel(Long policyId) {
+ return countActiveRows("policy_relation_meta", "policy_id = " + policyId);
+ }
+
private Integer countAllPolicyRel(Long policyId) {
try (SqlSession sqlSession =
SqlSessionFactoryHelper.getInstance().getSqlSessionFactory().openSession(true);
@@ -1078,4 +1581,40 @@ public class TestPolicyMetaService extends
TestJDBCBackend {
throw new RuntimeException("SQL execution failed", se);
}
}
+
+ private PolicyPO getPolicyPO(NameIdentifier identifier) {
+ return SessionUtils.getWithoutCommit(
+ PolicyMetaMapper.class,
+ mapper ->
+ mapper.selectPolicyMetaByMetalakeAndName(
+ identifier.namespace().level(0), identifier.name()));
+ }
+
+ /** Copies the policy under a new name and comment, keeping the audit info
tests create with. */
+ private PolicyEntity copyPolicy(PolicyEntity policy, String name, String
comment) {
+ return copyPolicy(policy, name, comment, AUDIT_INFO);
+ }
+
+ private PolicyEntity copyPolicy(
+ PolicyEntity policy, String name, String comment, AuditInfo auditInfo) {
+ return PolicyEntity.builder()
+ .withId(policy.id())
+ .withName(name)
+ .withNamespace(policy.namespace())
+ .withPolicyType(policy.policyType())
+ .withComment(comment)
+ .withEnabled(policy.enabled())
+ .withContent(policy.content())
+ .withAuditInfo(auditInfo)
+ .build();
+ }
+
+ private void await(CountDownLatch latch) {
+ try {
+ assertTrue(latch.await(30, TimeUnit.SECONDS));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ }
}
