jerryshao commented on code in PR #12456:
URL: https://github.com/apache/gravitino/pull/12456#discussion_r3841252170
##########
core/src/main/java/org/apache/gravitino/storage/relational/service/FilesetMetaService.java:
##########
@@ -166,6 +166,15 @@ public void insertFileset(FilesetEntity filesetEntity,
boolean overwrite) throws
// insert both fileset meta table and version table
SessionUtils.doMultipleWithCommit(
+ // Hold the parent schema row until this transaction ends, so the
fileset cannot be
+ // written below a schema that is being dropped.
+ () ->
+ SchemaMetaService.getInstance()
+ .lockSchemaForEntityWrite(
Review Comment:
**[Confirmed bug, surfaces at `FilesetCatalogOperations.java:575`]** When
this lock finds the schema gone (concurrently dropped),
`lockSchemaForEntityWrite` throws `NoSuchEntityException`. But the call site in
`FilesetCatalogOperations.createMultipleLocationFileset`'s final
`store.put(filesetEntity, true)` only catches `IOException`, so this exception
propagates uncaught — past the earlier schema-existence pre-check in that same
method, which correctly catches `NoSuchEntityException` and maps it to
`NoSuchSchemaException` (404).
**Failure scenario:** Client A calls `createFileset` under schema S; the
up-front `store.get(schemaIdent,...)` pre-check succeeds, then time passes
doing filesystem mkdir/validation work. Meanwhile client B cascade-drops schema
S. When A reaches `store.put(filesetEntity, true)`, this lock finds the schema
gone and throws `NoSuchEntityException`, which isn't
`NotFoundException`-derived and isn't special-cased in
`ExceptionHandlers.java`, so `doWithCatalog` lets it through unwrapped and the
REST layer returns a raw 500 instead of the intended 404. The analogous
managed-table path, `ManagedTableOperations.createTable`, does catch
`NoSuchEntityException` and convert it to `NoSuchSchemaException`, showing this
is the intended pattern the fileset call site misses.
##########
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:
**[Plausible efficiency]** `checkSchemaIsEmpty` materializes six full PO
lists (tables, filesets, models, topics, views, functions) via
`listXxxPOsBySchemaId` just to test `.isEmpty()`, while holding an exclusive
catalog-row lock the whole time.
A non-cascade drop of a schema with thousands of tables fetches and
deserializes every table/fileset/model/topic/view/function row under that
schema (six separate full scans) purely to answer a yes/no question, all while
every other schema-create/drop under the same catalog is blocked on the
exclusive catalog lock taken earlier in the same transaction. An existence
check (COUNT/LIMIT 1 or an exists-by-schema-id query) would answer the same
question with a fraction of the I/O and shorten how long the catalog stays
locked.
##########
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);
Review Comment:
**[Plausible simplification]** The "lock the parent schema row before
writing a child entity" pattern (this method) is hand-duplicated
near-identically in six call sites (`FilesetMetaService`,
`FunctionMetaService`, `ModelMetaService`, `ModelVersionMetaService`,
`TableMetaService`, `TopicMetaService`, `ViewMetaService`) instead of living
once in a shared write-path hook.
A future PR adding a seventh schema-scoped entity type could copy an
existing insert method without also adding the
`SchemaMetaService.getInstance().lockSchemaForEntityWrite(...)` call — nothing
in the type system or a shared base class forces it, so the new entity type
could silently reintroduce the exact orphan-under-a-dropped-schema race this PR
was written to close for every other type.
--
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]
