jerryshao commented on code in PR #12782:
URL: https://github.com/apache/gravitino/pull/12782#discussion_r3913958441
##########
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)
{
Review Comment:
**[correctness, CONFIRMED] Overwrite-by-name skips identity re-check after
lock**
`findAndLockPolicyForOverwrite` locks the resolved row by ID but never
re-checks it still has the name the overwrite was targeting, so a concurrent
rename can be silently reverted with no `OptimisticLockException`.
Concretely: row id=100 is named "P" (v5). T1 calls `insertPolicy(new entity
named "P", overwritten=true)`. `findAndLockPolicyForOverwrite` finds row 100
via an *unlocked* name lookup. Before T1 locks it, T2 renames row 100 "P"→"Q"
via `updatePolicy`, whose CAS succeeds (v5→v6). T1 then locks row 100 via
`selectPolicyByPolicyIdForUpdate(100)`, gets back name="Q"/v6, and — since
nothing checks the locked row's name still equals "P" — proceeds anyway. The
final CAS filters on `policy_id` + `current_version` only, not name, so it
trivially matches and commits, silently reverting T2's rename even though T2
committed after T1's initial observation. No test exercises a rename racing an
overwrite.
##########
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(
Review Comment:
**[correctness, CONFIRMED, low severity] Wrong identifier used in overwrite
CAS-failure message**
`insertPolicyWithoutCommit`'s overwrite branch passes the entity's *new*
`NameIdentifier` (rather than the locked row's actual observed identity) into
`updatePolicyRootWithVersion`.
When overwrite matches by policy ID with a different new name (row 100
currently named "P1", entity being written is id=100/name="P2" — a shape
exercised by `testPolicyOverwriteAdvancesVersionAndRetainsHistory`),
`updatePolicyRootWithVersion(policyEntity.nameIdentifier(), existingPolicyPO,
...)` is called with identifier="P2" while `existingPolicyPO`'s real name is
"P1". If the CAS ever misses, `policyWriteFailure` builds the exception message
using "P2" instead of "P1". The exception *type* is still correct (decided from
the locked row), so this is message-only, not a control-flow bug.
##########
core/src/main/java/org/apache/gravitino/storage/relational/service/PolicyMetaService.java:
##########
@@ -143,69 +141,82 @@ public <E extends Entity & HasIdentifier> PolicyEntity
updatePolicy(
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) {
Review Comment:
**[correctness, CONFIRMED] Concurrent double-delete now surfaces as HTTP
500**
`deletePolicy` now throws an unchecked `NoSuchEntityException` instead of
returning `false` when the row is deleted concurrently between the read and the
CAS delete.
Two nodes concurrently delete the same policy; node B commits first. Node
A's `deletePolicyWithVersion` → `OccWriteSupport.deleteWithVersion` re-checks
the row, finds it gone, and throws `NoSuchEntityException`.
`PolicyManager.deletePolicy` only catches `IOException`, so it propagates to
the REST layer's generic `catch(Exception)` → `ExceptionHandlers`, which has no
case for this on the delete path and falls through to `Utils.internalError` →
HTTP 500, ERROR-logged. The pre-PR implementation was a single atomic UPDATE
that always returned a clean boolean, consistent with `EntityStore.delete`'s
own javadoc contract ("@return ... false otherwise"). An idempotent "delete if
exists" caller now gets a 500 instead of a no-op.
##########
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()));
+ }
+
+ /**
+ * 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(
Review Comment:
**[altitude, PLAUSIBLE] Update-CAS idiom duplicated instead of centralized**
The update-CAS idiom (issue UPDATE, check affected-rows==0, classify via
re-lookup) is hand-rolled again here instead of being added to the shared
`OccWriteSupport` helper, which already centralizes the equivalent delete-CAS
idiom (`deleteWithVersion`).
`OccWriteSupport` exports `deleteWithVersion` and `lockParentForChildWrite`,
but no `updateWithVersion` counterpart. `TableMetaService` and
`TopicMetaService` already contain the identical inline pattern, and
`ModelVersionMetaService` a close variant; this PR adds a fourth copy rather
than lifting the pattern into `OccWriteSupport` once. A future concurrency fix
now has to be applied in four places instead of one.
##########
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) {
Review Comment:
**[reuse, CONFIRMED] lockMetalakeForPolicyCreate duplicates the catalog
version**
`lockMetalakeForPolicyCreate` is a byte-for-byte duplicate of
`CatalogMetaService.lockMetalakeForCatalogCreate` (same `OccWriteSupport` call,
same lookup, same predicate, same Javadoc) instead of a shared helper. Any
future fix to the "lock a metalake for a child-entity create" pattern must be
applied in both places.
##########
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);
Review Comment:
**[efficiency, PLAUSIBLE] JSON snapshot built while row lock is held**
The version-snapshot JSON serialization
(`POConverters.updatePolicyPOWithVersion`) runs after the row's exclusive lock
is already held in `findAndLockPolicyForOverwrite`, extending lock hold time
with CPU-bound work that doesn't need the lock. Under contention (many
concurrent overwrites of policies with large content payloads), every other
writer queues behind the row lock for the duration of serialization, not just
the actual UPDATE.
##########
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:
**[efficiency, PLAUSIBLE] 3 sequential SELECTs to resolve one overwrite
target**
`findAndLockPolicyForOverwrite` issues up to 3 sequential SELECTs to resolve
and lock one row (an always-missing ID lookup, an unlocked name lookup, then a
second locked ID lookup) where a single locked query would do. For the common
overwrite-by-name case the first `selectPolicyByPolicyIdForUpdate(newId)`
always misses since the new entity carries a freshly generated ID. A single
`SELECT ... WHERE metalake_id=? AND policy_name=? FOR UPDATE` would resolve and
lock the row in one round trip, and would also close the identity-recheck gap
noted above.
##########
core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyVersionMapper.java:
##########
@@ -44,6 +38,17 @@ void insertPolicyVersionOnDuplicateKeyUpdate(
Integer softDeletePolicyVersionByMetalakeAndPolicyName(
Review Comment:
**[simplification, CONFIRMED] Dead metalake+name-keyed delete mapper
methods**
`softDeletePolicyVersionByMetalakeAndPolicyName` and
`PolicyMetadataObjectRelMapper.softDeletePolicyMetadataObjectRelsByMetalakeAndPolicyName`
(plus their base/PostgreSQL SQL-provider implementations, ~6 locations) are
now dead: `deletePolicy` only calls the policy-ID-keyed variants. A repo-wide
grep finds zero callers outside their own declarations. The PR correctly
removed the sibling dead method `softDeletePolicyByMetalakeAndPolicyName` but
missed these two, leaving ~60 lines of unreachable code that will mislead a
future reader.
##########
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:
**[efficiency, PLAUSIBLE] Every policy alter now writes a version row
unconditionally**
`updatePolicyPOWithVersion` now unconditionally builds a new version-table
row on every alter; the old `checkPolicyVersionNeedUpdate` guard that skipped
the extra INSERT for metadata-only changes was removed. Confirmed intentional
and tested (`testMetadataOnlyPolicyAlterCreatesCompleteSnapshot` asserts a
version bump even when comment/enabled are identical), but it's real added
write/serialization cost on every alter call, including ones that previously
were no-ops on the version table.
##########
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:
**[simplification, PLAUSIBLE] New countActiveRows helper duplicates
countActivePolicyRel**
The new generic `countActiveRows(table, whereClause)` test helper duplicates
the pre-existing `countActivePolicyRel(policyId)`, which is just
`countActiveRows("policy_relation_meta", "policy_id = " + policyId)` inlined by
hand (~18 duplicated lines, identical error handling). Since this PR already
introduces the general form, `countActivePolicyRel` should have been rewritten
to call it.
--
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]
