jerryshao commented on code in PR #12782:
URL: https://github.com/apache/gravitino/pull/12782#discussion_r3949912799
##########
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:
**[test-coverage, PLAUSIBLE] Overwrite can clobber unrelated policy by name
collision**
`findAndLockPolicyForOverwrite` resolves the overwrite target by policy name
alone; it never checks that the caller-supplied policy ID matches the row found
by name, so `overwrite(entity)` can silently clobber a different, unrelated
existing policy that happens to hold the target name.
Concretely: two independent active policies exist in the same metalake,
A(id=1, name="alpha") and B(id=2, name="beta"). A caller invokes
`insertPolicy(entity, overwrite=true)` where `entity.id()==1` but
`entity.name()=="beta"`. This method finds B by name and returns it immediately
(the by-ID branch only runs on a name-miss). `insertPolicyWithoutCommit` then
overwrites B's content/audit-info with the payload intended for A, keeping B's
own `policy_id` — B's data is silently clobbered, A is untouched, no exception
is raised.
Verified this is an intentional repo-wide convention (the identical
by-name-first pattern exists in `TagMetaService.findAndLockTagForOverwrite`)
and currently latent: `overwrite=true` for policies is only exercised by unit
tests today, not reachable via `PolicyManager`'s REST path (always
`overwritten=false`). The existing test
`testPolicyOverwriteReplacesTheRowHoldingTheName` only covers overwriting with
a fresh, never-used ID — not the cross-ID case where the name is already owned
by a different real policy. Worth an explicit Javadoc callout and a test for
the cross-ID case before any future caller passes `overwrite=true`.
##########
core/src/main/java/org/apache/gravitino/storage/relational/service/OccWriteSupport.java:
##########
@@ -115,6 +115,20 @@ public static void deleteWithVersion(
}
}
+ /**
+ * Executes a single-row compare-and-set update for an entity guarded by
version.
+ *
+ * @param updateOps an operation supplying the number of rows affected by
the update
+ * @param onMissSupplier a supplier providing the RuntimeException when zero
rows are updated
+ */
+ public static void updateWithVersion(
Review Comment:
**[altitude, CONFIRMED] Update-CAS helper not backfilled to other services**
The new shared `updateWithVersion(IntSupplier, Supplier<RuntimeException>)`
helper — a genuine, well-built counterpart to the existing `deleteWithVersion`
— is used only by `PolicyMetaService`; `TableMetaService`, `TopicMetaService`,
`CatalogMetaService`, and `ModelVersionMetaService` still hand-roll the
identical "issue UPDATE, check affected-rows==0, classify via re-lookup"
pattern inline, unmigrated.
`TableMetaService.java:241`, `TopicMetaService.java:135`,
`CatalogMetaService.java:253`, and `ModelVersionMetaService.java:392/243` all
keep their own copy of the exact idiom this PR just centralized for Policy. Not
a functional bug today (both forms are currently semantically identical), but
it leaves five copies of the same CAS-classification logic in the codebase
instead of one — a future fix to the CAS-retry/error-classification behavior
applied only to `OccWriteSupport.updateWithVersion` will silently miss
Table/Topic/Catalog/ModelVersion, reintroducing the exact class of bug this PR
set out to eliminate.
##########
core/src/main/java/org/apache/gravitino/storage/relational/service/PolicyMetaService.java:
##########
@@ -93,36 +103,35 @@ public void insertPolicy(PolicyEntity policyEntity,
boolean overwritten) throws
String metalakeName = ns.level(0);
try {
- Long metalakeId =
- EntityIdService.getEntityId(NameIdentifier.of(metalakeName),
Entity.EntityType.METALAKE);
+ MetalakePO metalakePO =
Review Comment:
**[reuse, PLAUSIBLE] Metalake-lookup-by-name block duplicated across
services**
The "look up metalake by name, throw `NoSuchEntityException` if missing"
block in `insertPolicy` duplicates the identical block in
`CatalogMetaService.insertCatalog`.
Both blocks select via `MetalakeMetaMapper.selectMetalakeMetaByName`,
null-check, and throw the identical `NoSuchEntityException` with
`Entity.EntityType.METALAKE`. Combined with the lock-method duplication noted
separately, a single shared helper (e.g. `requireMetalakeByName(String)`)
returning a validated `MetalakePO` would cut ~9 duplicated lines per call site,
currently duplicated at least twice in this codebase.
##########
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) {
Review Comment:
**[reuse, CONFIRMED] lockMetalakeForPolicyCreate duplicates catalog's
version**
`lockMetalakeForPolicyCreate` remains a byte-for-byte duplicate of
`CatalogMetaService.lockMetalakeForCatalogCreate` (identical body and Javadoc)
— unaddressed since the prior review round on this same PR.
Both methods call `OccWriteSupport.lockParentForChildWrite` with the same
lookup and identity predicate, word-for-word. A third near-identical copy
(`lockCatalogForSchemaCreate`) exists in `SchemaMetaService`. Any future fix to
the "lock parent for child create" pattern must be applied in multiple places
by hand.
##########
core/src/main/java/org/apache/gravitino/storage/relational/service/PolicyMetaService.java:
##########
@@ -143,69 +152,87 @@ 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:
**[efficiency, PLAUSIBLE] Delete pulls full policy content just for 2
scalars**
`deletePolicy` now routes through `getPolicyPOByMetalakeAndName`, a 3-table
JOIN (policy_meta + metalake_meta + policy_version_info) that pulls the full
JSON `content` column, just to obtain the `policyId` + `currentVersion` the CAS
soft-delete actually needs.
Every policy delete now transfers a full policy-content JSON blob and joins
two extra tables that the old single-UPDATE-by-name implementation never
touched, for a delete path that only needs two scalar columns to perform its
CAS. On a policy with large content, this is pure wasted I/O and network
transfer on every delete call.
##########
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:
**[simplification, PLAUSIBLE] listPolicyPOsByPolicyIdsForUpdate copy-pastes
base SQL**
`listPolicyPOsByPolicyIdsForUpdate` re-implements the full 19-line SQL body
of `listPolicyPOsByPolicyIds` instead of delegating to it and appending `FOR
UPDATE`, unlike the sibling method `selectPolicyByPolicyIdForUpdate` which
correctly builds on its base query (`return selectPolicyByPolicyId(id) + " FOR
UPDATE"`).
`listPolicyPOsByPolicyIdsForUpdate` instead copy-pastes the entire
`<script>...SELECT...FOR UPDATE</script>` block from
`listPolicyPOsByPolicyIds`, so any future column change to the base query must
be edited in two places, and the two can silently drift out of sync.
##########
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(
Review Comment:
**[altitude, PLAUSIBLE] Overwrite conflict checks hand-rolled instead of
shared**
The overwrite path's pre-insert conflict classification (reserved-deleted-ID
reuse here, cross-metalake ID reuse elsewhere in this method) is hand-rolled as
ad-hoc inline `EntityAlreadyExistsException` throws rather than routed through
`OccWriteSupport`'s existing "lookup + identity predicate + classify" mechanism.
These checks were added piecemeal across the "polish" and "classify
conflicts safely" review-round commits, specific to the policy overwrite-by-ID
path. A future entity type needing the same "ID reuse after delete" or
"cross-parent ID reuse" checks will re-derive the same throw-inline idiom from
scratch instead of calling a generalized version of
`OccWriteSupport.writeFailure`/`lockParentForChildWrite` extended to cover
pre-insert conflict classification.
##########
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()))
Review Comment:
**[efficiency, PLAUSIBLE] Overwrite-miss path does 3 sequential queries**
The overwrite-miss path in
`insertPolicyWithoutCommit`/`findAndLockPolicyForOverwrite` issues 3 sequential
SELECTs (locked-by-name miss, locked-by-id lookup, then this
`countDeletedPolicyMetasById` reservation check) where folding the `deleted_at`
check into the second query's predicate would need only 2.
This third query exists only because the second query filters `deleted_at =
0`, hiding a soft-deleted row with the same ID. Checking `deletedAt` in Java
instead of filtering it out in SQL would fold the "reserved by a deleted
policy" check into the same round trip, saving one query per
overwrite-of-a-new-name.
##########
core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyMetaMapper.java:
##########
@@ -146,6 +161,29 @@ Integer deletePolicyMetasByLegacyTimeline(
PolicyPO selectPolicyMetaByMetalakeIdAndName(
@Param("metalakeId") long metalakeId, @Param("policyName") String
policyName);
+ /**
+ * Selects and exclusively locks an active policy by its natural key.
+ *
+ * @param metalakeId The metalake ID.
+ * @param policyName The policy name.
+ * @return The locked policy, or null if the natural key is not active.
+ */
+ @Results({
Review Comment:
**[simplification, PLAUSIBLE] Repeated @Results blocks instead of shared
@ResultMap**
Ten identical 8-line `@Results({...})` blocks mapping the same 8 `PolicyPO`
columns are repeated verbatim across the mapper (3 of them newly added by this
PR for the `*ForUpdate` query variants), instead of using this codebase's own
`@ResultMap(id=...)` reuse convention.
`FunctionMetaMapper.java` and `ViewMetaMapper.java` elsewhere in this
codebase already declare one `@Results(id = "...")` and reference it from other
methods via `@ResultMap("...")`. `PolicyMetaMapper` doesn't follow that
convention, so every new "for update" variant this PR adds pays another 8-line
copy instead of one annotation, and a future column addition/rename must be
replicated across all 10 call sites by hand.
##########
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:
**[conventions, CONFIRMED] New private helpers still misplaced among public
methods**
A new private helper `associatePoliciesWithMetadataObjectWithoutCommit` is
inserted between two public methods, and a ~230-line block of new
private/package-private helpers is inserted before the pre-existing public
methods `getPolicyIdByPolicyName` and `batchGetPolicyByIdentifier` at the tail
of the class — the same class-member-ordering issue flagged in the prior review
round on this PR, still unaddressed.
Per this repo's CLAUDE.md: "Methods (Group by visibility, putting private
methods at the end)." The new private helper sits between
`associatePoliciesWithMetadataObject` (public) and
`deletePolicyAndVersionMetasByLegacyTimeline` (public), and the large new
private-helpers block precedes two pre-existing public methods at the class
tail — private methods are not grouped at the end as the rule requires, exactly
as in the prior review round.
--
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]
