yuqi1129 commented on code in PR #12782:
URL: https://github.com/apache/gravitino/pull/12782#discussion_r3904186490
##########
core/src/test/java/org/apache/gravitino/storage/relational/service/TestPolicyMetaService.java:
##########
@@ -1078,4 +1229,26 @@ private Integer countAllPolicyRel(Long policyId) {
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()));
+ }
+
+ private PolicyEntity copyPolicy(
+ PolicyEntity policy, String name, String comment, Audit 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) auditInfo)
+ .build();
Review Comment:
Fixed. `copyPolicy` takes `AuditInfo` and the cast is gone; the callers that
only rename or re-comment use a new three-argument overload that reuses the
tests' `AUDIT_INFO`.
##########
core/src/main/java/org/apache/gravitino/storage/relational/utils/POConverters.java:
##########
@@ -761,58 +760,35 @@ public static FilesetPO updateFilesetPOWithVersion(
}
}
- public static boolean checkPolicyVersionNeedUpdate(
- PolicyVersionPO oldPolicyVersionPO, PolicyEntity newPolicy) {
- if (!StringUtils.equals(oldPolicyVersionPO.getPolicyComment(),
newPolicy.comment())
- || oldPolicyVersionPO.isEnabled() != newPolicy.enabled()) {
- return true;
- }
-
- try {
- PolicyContent oldContent =
- JsonUtils.anyFieldMapper()
- .readValue(oldPolicyVersionPO.getContent(),
newPolicy.policyType().contentClass());
- if (oldContent == null) {
- return newPolicy.content() != null;
- }
- return !oldContent.equals(newPolicy.content());
- } catch (JsonProcessingException e) {
- throw new RuntimeException("Failed to deserialize json object:", e);
- }
- }
-
- public static PolicyPO updatePolicyPOWithVersion(
- PolicyPO oldPolicyPO, PolicyEntity newPolicy, boolean needUpdateVersion)
{
+ /**
+ * Builds the next complete policy metadata and content snapshot.
+ *
+ * @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 {
- 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();
- }
+ Long nextVersion =
+ Math.max(oldPolicyPO.getCurrentVersion(),
oldPolicyPO.getLastVersion()) + 1;
+ PolicyVersionPO newPolicyVersionPO =
+ PolicyVersionPO.builder()
+ .withMetalakeId(oldPolicyPO.getMetalakeId())
+ .withPolicyId(oldPolicyPO.getPolicyId())
+ .withVersion(nextVersion)
+ .withPolicyComment(newPolicy.comment())
+ .withEnabled(newPolicy.enabled())
+
.withContent(JsonUtils.anyFieldMapper().writeValueAsString(newPolicy.content()))
+ .withDeletedAt(DEFAULT_DELETED_AT)
+ .build();
return PolicyPO.builder()
- .withPolicyId(newPolicy.id())
+ .withPolicyId(oldPolicyPO.getPolicyId())
.withPolicyName(newPolicy.name())
.withPolicyType(newPolicy.policyType().policyType())
.withMetalakeId(oldPolicyPO.getMetalakeId())
.withAuditInfo(JsonUtils.anyFieldMapper().writeValueAsString(newPolicy.auditInfo()))
- .withCurrentVersion(currentVersion)
- .withLastVersion(lastVersion)
+ .withCurrentVersion(nextVersion)
+ .withLastVersion(nextVersion)
Review Comment:
Deliberate, and now documented on the method. `oldPolicyPO` is the row being
replaced, and its ID is what every version snapshot and relation row points at,
so the row keeps it.
An alter cannot change the ID: `PolicyMetaService.updatePolicy` rejects an
updater that returns a different one via `Preconditions`. An overwrite of a
name held by another row updates that row rather than inserting a second one
under the same name, so the caller-supplied ID is dropped on purpose;
`testPolicyOverwriteReplacesTheRowHoldingTheName` pins that behaviour.
##########
core/src/main/java/org/apache/gravitino/storage/relational/service/PolicyMetaService.java:
##########
@@ -440,6 +445,142 @@ public int deletePolicyVersionsByRetentionCount(Long
versionRetentionCount, int
return totalDeletedCount;
}
+ 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()));
+ }
+
+ private void insertPolicyWithoutCommit(
+ PolicyEntity policyEntity, PolicyPO initializedPolicyPO, boolean
overwritten) {
+ if (!overwritten) {
+ insertNewPolicyWithoutCommit(initializedPolicyPO);
+ return;
+ }
+
+ PolicyPO existingPolicyPO =
findAndLockPolicyForOverwrite(initializedPolicyPO);
+ if (existingPolicyPO == null) {
+ insertNewPolicyWithoutCommit(initializedPolicyPO);
+ return;
+ }
+
+ PolicyPO replacementPolicyPO =
+ POConverters.updatePolicyPOWithVersion(existingPolicyPO, policyEntity);
+ updatePolicyRootWithVersion(
+ policyEntity.nameIdentifier(), 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()));
+ }
+
+ private PolicyPO findAndLockPolicyForOverwrite(PolicyPO initializedPolicyPO)
{
+ PolicyPO existingPolicyPO =
+ SessionUtils.getWithoutCommit(
+ PolicyMetaMapper.class,
+ mapper ->
mapper.selectPolicyByPolicyIdForUpdate(initializedPolicyPO.getPolicyId()));
+ if (existingPolicyPO != null) {
+ return existingPolicyPO;
+ }
+
+ PolicyPO sameNamePolicyPO =
+ SessionUtils.getWithoutCommit(
+ PolicyMetaMapper.class,
+ mapper ->
+ mapper.selectPolicyMetaByMetalakeIdAndName(
+ initializedPolicyPO.getMetalakeId(),
initializedPolicyPO.getPolicyName()));
+ if (sameNamePolicyPO == null) {
+ return null;
+ }
+ return SessionUtils.getWithoutCommit(
+ PolicyMetaMapper.class,
+ mapper ->
mapper.selectPolicyByPolicyIdForUpdate(sameNamePolicyPO.getPolicyId()));
+ }
+
+ private void updatePolicyRootWithVersion(
+ NameIdentifier identifier, PolicyPO oldPolicyPO, PolicyPO newPolicyPO) {
+ int updated =
+ SessionUtils.getWithoutCommit(
+ PolicyMetaMapper.class, mapper ->
mapper.updatePolicyMeta(newPolicyPO, oldPolicyPO));
+ if (updated == 0) {
+ throw 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()));
+ }
+
+ private Map<Long, PolicyPO> lockPoliciesForAssociation(
+ List<PolicyPO> policyPOsToAdd, List<PolicyPO> policyPOsToRemove) {
+ Map<Long, PolicyPO> observedPolicyPOs = new LinkedHashMap<>();
+ policyPOsToAdd.forEach(policyPO ->
observedPolicyPOs.put(policyPO.getPolicyId(), policyPO));
+ policyPOsToRemove.forEach(policyPO ->
observedPolicyPOs.put(policyPO.getPolicyId(), policyPO));
+ List<PolicyPO> sortedPolicyPOs = new
ArrayList<>(observedPolicyPOs.values());
+ sortedPolicyPOs.sort(Comparator.comparingLong(PolicyPO::getPolicyId));
+
+ Map<Long, PolicyPO> lockedPolicyPOs = new LinkedHashMap<>();
+ for (PolicyPO observedPolicyPO : sortedPolicyPOs) {
+ PolicyPO lockedPolicyPO =
+ SessionUtils.getWithoutCommit(
+ PolicyMetaMapper.class,
+ mapper ->
mapper.selectPolicyByPolicyIdForUpdate(observedPolicyPO.getPolicyId()));
+ if (lockedPolicyPO == null
+ || !Objects.equals(lockedPolicyPO.getPolicyName(),
observedPolicyPO.getPolicyName())
+ || !Objects.equals(lockedPolicyPO.getMetalakeId(),
observedPolicyPO.getMetalakeId())) {
+ throw new NoSuchEntityException(
+ NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
+ Entity.EntityType.POLICY.name().toLowerCase(),
+ observedPolicyPO.getPolicyName());
+ }
+ lockedPolicyPOs.put(lockedPolicyPO.getPolicyId(), lockedPolicyPO);
+ }
+ return lockedPolicyPOs;
Review Comment:
Fixed. `PolicyMetaService.lockPolicies` now issues one
`listPolicyPOsByPolicyIdsForUpdate` (`WHERE policy_id IN (...) ORDER BY
policy_id FOR UPDATE`) and validates the locked rows against the observed set,
so an association change touching N policies costs one round trip with the same
policy-ID lock order. `PolicyTagRelService` uses the same helper.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
