yuqi1129 commented on code in PR #12456:
URL: https://github.com/apache/gravitino/pull/12456#discussion_r3841302323
##########
core/src/main/java/org/apache/gravitino/storage/relational/service/SchemaMetaService.java:
##########
@@ -450,12 +449,196 @@ private List<SchemaPO> listSchemaPOs(Namespace
namespace) {
}
/**
- * Collects the schema ids that participate in a cascade delete: the target
schema itself plus
- * every HierarchicalSchema descendant. The {@link SchemaPO} arrives in
logical form (e.g. {@code
- * A:B}); {@link HierarchicalConversionPOStorageOps} translates to storage
form before running the
- * SQL prefix match, so this method only deals in logical names.
+ * Holds the parent catalog row for the rest of the transaction, so a schema
cannot be created
+ * below a catalog that is being dropped. Dropping a catalog locks this same
row, so the two can
+ * never run at the same time: the loser either finds the catalog gone or
inserts below a catalog
+ * that is still there.
+ *
+ * <p>A plain schema name only needs a shared lock, so many schemas can be
created under one
+ * catalog at once. A nested name is different: this request may have to
create the missing
+ * ancestors, and two requests can both find the same ancestor missing and
both insert it. A
+ * shared lock does not stop that, so the ancestor case takes an exclusive
lock and serializes
+ * every other schema create under the catalog until it finishes.
+ *
+ * <p>The name and the metalake are compared again because the caller looked
the catalog up by
+ * name: if the row now has another name, the catalog named in the request
no longer exists.
+ */
+ 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());
+ }
+ }
+
+ /**
+ * Holds the parent catalog row while a schema is dropped. The lock is
exclusive here, because a
+ * drop removes descendants and must not run next to another drop or create
under the same
+ * catalog. Taking the catalog before any schema row also gives every drop
the same lock order, so
+ * two overlapping cascades cannot deadlock.
*/
- private List<Long> listSchemaIdsForCascade(SchemaPO schemaPO) {
+ 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));
+ }
+ }
+
+ /**
+ * Holds the parent schema row while a table, view, fileset, function,
model, model version, or
+ * topic is written, so a child cannot be added below a schema that is going
away. The lock is
+ * shared, so children of the same schema can still be written in parallel;
dropping the schema
+ * takes the row exclusively and therefore waits for them.
+ */
+ void lockSchemaForEntityWrite(
+ NameIdentifier entityIdentifier,
+ Long observedSchemaId,
+ Long observedCatalogId,
+ Long observedMetalakeId) {
+ NameIdentifier schemaIdentifier =
NameIdentifierUtil.getSchemaIdentifier(entityIdentifier);
+ SchemaPO currentSchemaPO =
+ SessionUtils.getWithoutCommit(
+ SchemaMetaMapper.class,
+ mapper -> mapper.selectSchemaMetaByIdForShare(observedSchemaId));
+ if (currentSchemaPO != null) {
+ currentSchemaPO = physicalToLogicalSchemaPO(currentSchemaPO);
+ }
+ if (currentSchemaPO == null
+ || !Objects.equals(currentSchemaPO.getSchemaName(),
schemaIdentifier.name())
+ || !Objects.equals(currentSchemaPO.getCatalogId(), observedCatalogId)
+ || !Objects.equals(currentSchemaPO.getMetalakeId(),
observedMetalakeId)) {
+ throw noSuchSchemaException(schemaIdentifier);
+ }
+ }
+
+ /**
+ * Decides which error a failed compare-and-set should report. The write
matched no row either
+ * because somebody else changed the schema, which is a conflict, or because
the schema was
+ * deleted or renamed away, which is a missing entity.
+ */
+ private RuntimeException schemaWriteFailure(
+ NameIdentifier identifier, SchemaPO observedSchemaPO) {
+ // Sessions run at READ_COMMITTED, so a plain read would already see the
latest committed row.
+ // The locking read additionally waits for a writer that is still in
flight, so a delete or
+ // rename that has not committed yet is reported as a missing schema
instead of as a stale
+ // version conflict. The lock is taken on the error path of a transaction
that is about to roll
+ // back.
+ SchemaPO currentSchemaPO =
+ SessionUtils.getWithoutCommit(
+ SchemaMetaMapper.class,
+ mapper ->
mapper.selectSchemaMetaByIdForUpdate(observedSchemaPO.getSchemaId()));
+ if (currentSchemaPO == null) {
+ return noSuchSchemaException(identifier);
+ }
+ currentSchemaPO = physicalToLogicalSchemaPO(currentSchemaPO);
+ if (!Objects.equals(currentSchemaPO.getSchemaName(),
observedSchemaPO.getSchemaName())
+ || !Objects.equals(currentSchemaPO.getCatalogId(),
observedSchemaPO.getCatalogId())
+ || !Objects.equals(currentSchemaPO.getMetalakeId(),
observedSchemaPO.getMetalakeId())) {
+ return noSuchSchemaException(identifier);
+ }
+ return ExceptionUtils.concurrentModification(Entity.EntityType.SCHEMA,
identifier);
+ }
+
+ private NoSuchEntityException noSuchSchemaException(NameIdentifier
identifier) {
+ return new NoSuchEntityException(
+ NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
+ Entity.EntityType.SCHEMA.name().toLowerCase(),
+ identifier.name());
+ }
+
+ /**
+ * Soft-deletes the nested schemas below the dropped one, each guarded by
the version read in the
+ * same transaction.
+ */
+ private void deleteDescendantSchemasWithVersions(
+ NameIdentifier schemaIdentifier, List<SchemaPO> descendants) {
+ if (descendants.isEmpty()) {
+ return;
+ }
+ int deleted =
+ SessionUtils.getWithoutCommit(
+ SchemaMetaMapper.class, mapper ->
mapper.softDeleteSchemaMetasWithVersion(descendants));
+ // A smaller count means one of these schemas was altered by a request
that did not take the
+ // catalog lock. Never commit half a cascade: roll the whole transaction
back instead.
+ if (deleted != descendants.size()) {
+ throw ExceptionUtils.concurrentChildModification(
+ Entity.EntityType.SCHEMA, Entity.EntityType.SCHEMA,
schemaIdentifier);
+ }
+ }
+
+ /**
+ * Checks that nothing is left under the schema. Views and functions are
included: they used to be
+ * missing here, which let a non-cascade drop leave their rows behind with
no parent.
+ */
+ private void checkSchemaIsEmpty(NameIdentifier identifier, SchemaPO
schemaPO) {
+ boolean hasDescendantSchemas =
!listDescendantSchemaPOs(schemaPO).isEmpty();
Review Comment:
Yeah, It's indeed a performance problem. I would not modify it as this PR is
already too large, and fixing it will introuce at least 300 lines of code
changes. Moreover, the problem has existed for a long time; I would suggest
fixing it in another PR.
--
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]
