Copilot commented on code in PR #12782:
URL: https://github.com/apache/gravitino/pull/12782#discussion_r3902557825
##########
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:
`updatePolicyPOWithVersion` intentionally persists the *existing* stable
`policy_id` (from `oldPolicyPO`) and ignores `newPolicy.id()`. If callers can
supply a different ID during overwrite/alter flows, this can lead to surprising
behavior (persisted row uses old ID while the provided entity carries a
different one). Consider enforcing an invariant (e.g., validate
`newPolicy.id()` matches `oldPolicyPO.getPolicyId()` and fail fast) or
explicitly normalizing the incoming entity to the stable ID at the service
layer.
##########
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:
This performs one `SELECT ... FOR UPDATE` per policy (N+1 round trips)
during association updates, which can become a noticeable bottleneck when many
policies are added/removed. Consider adding a mapper method that locks all
required policy IDs in one statement (e.g., `WHERE policy_id IN (...) ORDER BY
policy_id FOR UPDATE`) and then validating the returned rows against the
observed set; this keeps the deadlock-avoidance ordering while reducing query
overhead.
##########
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:
The helper takes `Audit` but then unconditionally casts to `AuditInfo`,
which is an unsafe runtime cast (a different `Audit` implementation would cause
a `ClassCastException`). In tests, prefer taking `AuditInfo` directly (or
removing the cast by aligning the parameter type with
`PolicyEntity.builder().withAuditInfo(...)`).
--
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]
