yuqi1129 commented on code in PR #12782:
URL: https://github.com/apache/gravitino/pull/12782#discussion_r3914478480
##########
core/src/main/java/org/apache/gravitino/storage/relational/service/PolicyMetaService.java:
##########
@@ -440,6 +459,205 @@ public int deletePolicyVersionsByRetentionCount(Long
versionRetentionCount, int
return totalDeletedCount;
}
+ /**
+ * 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; a create of the same name that commits in between
then surfaces as an
+ * already-exists failure that the caller retries, which is the same outcome
the caller would see
+ * had the two requests arrived in the other order.
+ *
+ * <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) {
+ 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()));
Review Comment:
Fixed in 3a4f768 together with the correctness issue. The common
overwrite-by-name path is now one locking natural-key query. A same-ID rename
uses the natural-key miss followed by the stable-ID lock, so the uncommon
rename path uses at most two queries instead of three.
##########
core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyVersionMapper.java:
##########
@@ -44,6 +38,17 @@ void insertPolicyVersionOnDuplicateKeyUpdate(
Integer softDeletePolicyVersionByMetalakeAndPolicyName(
Review Comment:
Fixed in 3a4f768. Removed both dead natural-key cleanup APIs from the
mapper, provider factory, base provider, and PostgreSQL provider surfaces. A
repo-wide search now finds no declarations or callers.
##########
core/src/main/java/org/apache/gravitino/storage/relational/utils/POConverters.java:
##########
@@ -761,58 +760,41 @@ 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.
+ *
+ * <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) {
Review Comment:
Kept intentionally because it is a correctness requirement of #12776, not
only a history choice: every successful alter must advance `current_version`,
including metadata-only changes. If such an update reused the old token, a
second writer with the same observed version could still match and silently
overwrite it. The complete snapshot is required because reads join the root
current version to `policy_version_info`;
`testMetadataOnlyPolicyAlterCreatesCompleteSnapshot` pins this invariant.
##########
core/src/test/java/org/apache/gravitino/storage/relational/service/TestPolicyMetaService.java:
##########
@@ -1040,6 +1323,24 @@ private EntitiesToTest createAndAssociateEntities(
return new EntitiesToTest(catalog, schema, table, topic, fileset, model);
}
+ private int countActiveRows(String table, String whereClause) {
Review Comment:
Fixed in 3a4f768. `countActivePolicyRel` now delegates to
`countActiveRows("policy_relation_meta", ...)`; the duplicate JDBC
resource/error-handling block is gone.
--
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]
