Copilot commented on code in PR #12350:
URL: https://github.com/apache/gravitino/pull/12350#discussion_r3748836456
##########
core/src/main/java/org/apache/gravitino/storage/relational/service/SchemaMetaService.java:
##########
@@ -492,13 +470,151 @@ private List<SchemaPO> listSchemaPOs(Namespace
namespace) {
mapper -> POStorageReadRouting.listPOs(mapper, namespace, ops,
Entity.EntityType.SCHEMA));
}
+ private void lockCatalogForSchemaCreate(
+ CatalogPO observedCatalogPO, boolean createsImplicitAncestors) {
+ CatalogPO currentCatalogPO =
+ SessionUtils.getWithoutCommit(
+ CatalogMetaMapper.class,
+ mapper ->
+ createsImplicitAncestors
+ ?
mapper.selectCatalogMetaByIdForUpdate(observedCatalogPO.getCatalogId())
+ :
mapper.selectCatalogMetaByIdForShare(observedCatalogPO.getCatalogId()));
+ if (currentCatalogPO == null
+ || !Objects.equals(currentCatalogPO.getCatalogName(),
observedCatalogPO.getCatalogName())
+ || !Objects.equals(currentCatalogPO.getMetalakeId(),
observedCatalogPO.getMetalakeId())) {
+ throw new NoSuchEntityException(
+ NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
+ Entity.EntityType.CATALOG.name().toLowerCase(),
+ observedCatalogPO.getCatalogName());
+ }
+ }
+
+ private void lockCatalogForSchemaDelete(NameIdentifier identifier, SchemaPO
observedSchemaPO) {
+ CatalogPO currentCatalogPO =
+ SessionUtils.getWithoutCommit(
+ CatalogMetaMapper.class,
+ mapper ->
mapper.selectCatalogMetaByIdForUpdate(observedSchemaPO.getCatalogId()));
+ if (currentCatalogPO == null
+ || !Objects.equals(currentCatalogPO.getCatalogName(),
identifier.namespace().level(1))
+ || !Objects.equals(currentCatalogPO.getMetalakeId(),
observedSchemaPO.getMetalakeId())) {
+ throw new NoSuchEntityException(
+ NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
+ Entity.EntityType.CATALOG.name().toLowerCase(),
+ identifier.namespace().level(1));
+ }
+ }
+
+ void lockSchemaForEntityWrite(
+ NameIdentifier entityIdentifier,
+ Long observedSchemaId,
+ Long observedCatalogId,
+ Long observedMetalakeId) {
Review Comment:
This fence is currently used only by child inserts and cross-schema table
moves, so it does not prevent all writes after a schema cascade. For example,
`updateFunction` can read the function before the cascade, then—after the
cascade commits—insert a new function-version row and ignore the zero-row
metadata update, committing an active orphan; same-schema `updateTable`
similarly commits its new version before reporting failure, and
`insertModelVersion` has no schema fence. Acquire this shared schema lock at
the start of every child write transaction (and fail inside the transaction
when the child CAS/update misses), not only on inserts/moves.
--
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]
