yuqi1129 commented on code in PR #12782:
URL: https://github.com/apache/gravitino/pull/12782#discussion_r3950168081
##########
core/src/main/java/org/apache/gravitino/storage/relational/service/PolicyMetaService.java:
##########
@@ -313,81 +340,89 @@ public List<PolicyEntity>
associatePoliciesWithMetadataObject(
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;
}
}
+ private List<PolicyEntity> associatePoliciesWithMetadataObjectWithoutCommit(
Review Comment:
Fixed in 13a424f1. `getPolicyIdByPolicyName` and
`batchGetPolicyByIdentifier` moved above the private block, and
`associatePoliciesWithMetadataObjectWithoutCommit` moved down into it, so the
class is public-then-private again. Pure movement, no logic change.
##########
core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/PolicyMetaBaseSQLProvider.java:
##########
@@ -185,6 +160,34 @@ public String selectPolicyByPolicyId(@Param("policyId")
Long policyId) {
+ " AND pm.deleted_at = 0 ";
}
+ /** 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) {
Review Comment:
Fixed in 13a424f1. Both by-ID list queries now wrap a shared
`selectPolicyPOsByPolicyIdsBody()`, with the locking variant appending `ORDER
BY pm.policy_id FOR UPDATE`.
One note on the suggested idiom: `base + " FOR UPDATE"` (as
`selectPolicyByPolicyIdForUpdate` does) does not work for these two, because
they are `<script>`-wrapped and the clause has to go inside the tag. Extracting
the body is the equivalent fix. `PolicyMetaPostgreSQLProvider` overrides
neither method, so the change is contained to the base provider.
##########
core/src/main/java/org/apache/gravitino/storage/relational/service/PolicyMetaService.java:
##########
@@ -440,6 +475,217 @@ 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. 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()));
+ }
+
+ private PolicyPO findAndLockPolicyForOverwrite(PolicyPO initializedPolicyPO)
{
Review Comment:
Documented rather than changed, in 13a424f1.
`findAndLockPolicyForOverwrite` now states that the name is the primary key
of the search, and that a caller supplying one policy's ID together with
another policy's name replaces the row holding the name. As the comment notes,
this matches `TagMetaService.findAndLockTagForOverwrite`, and
`PolicyManager.java:165` is the only production writer — it always passes
`overwritten = false`, so the path is reachable from tests only.
I would rather not add the cross-ID test: it would promote a currently
unreachable behaviour into a pinned contract that has to be maintained. The
Javadoc is the right level of commitment until a caller actually needs
`overwrite = true`.
--
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]
