diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/SchemaMetaMapper.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/SchemaMetaMapper.java index bb870e44dc4..1f6c844c191 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/SchemaMetaMapper.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/SchemaMetaMapper.java @@ -86,6 +86,16 @@ SchemaPO selectSchemaByFullQualifiedName( @SelectProvider(type = SchemaMetaSQLProviderFactory.class, method = "selectSchemaMetaById") SchemaPO selectSchemaMetaById(@Param("schemaId") Long schemaId); + /** + * Returns one when an active table, view, fileset, function, model, or topic exists in the + * schema, and {@code null} otherwise. + * + *

Only a literal is selected because callers need an existence answer, not complete child + * metadata. The final limit also lets the database stop as soon as it finds the first child. + */ + @SelectProvider(type = SchemaMetaSQLProviderFactory.class, method = "selectActiveChildBySchemaId") + Integer selectActiveChildBySchemaId(@Param("schemaId") Long schemaId); + /** Selects and locks an active schema by ID for the current transaction. */ @SelectProvider( type = SchemaMetaSQLProviderFactory.class, diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/SchemaMetaSQLProviderFactory.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/SchemaMetaSQLProviderFactory.java index 62c532db549..bcebb42eb92 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/SchemaMetaSQLProviderFactory.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/SchemaMetaSQLProviderFactory.java @@ -106,6 +106,11 @@ public static String selectSchemaMetaById(@Param("schemaId") Long schemaId) { return getProvider().selectSchemaMetaById(schemaId); } + /** Returns SQL that checks whether an active child exists in the schema. */ + public static String selectActiveChildBySchemaId(@Param("schemaId") Long schemaId) { + return getProvider().selectActiveChildBySchemaId(schemaId); + } + /** Returns SQL that selects and locks an active schema by ID. */ public static String selectSchemaMetaByIdForUpdate(@Param("schemaId") Long schemaId) { return getProvider().selectSchemaMetaByIdForUpdate(schemaId); diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/SchemaMetaBaseSQLProvider.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/SchemaMetaBaseSQLProvider.java index 76f989b3b6b..b5966f07ae1 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/SchemaMetaBaseSQLProvider.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/SchemaMetaBaseSQLProvider.java @@ -22,7 +22,13 @@ import java.util.List; import org.apache.gravitino.storage.relational.mapper.CatalogMetaMapper; +import org.apache.gravitino.storage.relational.mapper.FilesetMetaMapper; +import org.apache.gravitino.storage.relational.mapper.FunctionMetaMapper; import org.apache.gravitino.storage.relational.mapper.MetalakeMetaMapper; +import org.apache.gravitino.storage.relational.mapper.ModelMetaMapper; +import org.apache.gravitino.storage.relational.mapper.TableMetaMapper; +import org.apache.gravitino.storage.relational.mapper.TopicMetaMapper; +import org.apache.gravitino.storage.relational.mapper.ViewMetaMapper; import org.apache.gravitino.storage.relational.po.SchemaPO; import org.apache.ibatis.annotations.Param; @@ -182,6 +188,31 @@ public String selectSchemaMetaById(@Param("schemaId") Long schemaId) { + " WHERE schema_id = #{schemaId} AND deleted_at = 0"; } + /** Returns SQL that checks whether an active child exists in the schema. */ + public String selectActiveChildBySchemaId(@Param("schemaId") Long schemaId) { + // Each branch returns only the same literal, so UNION ALL avoids unnecessary duplicate + // elimination. LIMIT 1 lets the database stop as soon as any kind of child is found. + return "SELECT 1 FROM " + + TableMetaMapper.TABLE_NAME + + " WHERE schema_id = #{schemaId} AND deleted_at = 0" + + " UNION ALL SELECT 1 FROM " + + ViewMetaMapper.TABLE_NAME + + " WHERE schema_id = #{schemaId} AND deleted_at = 0" + + " UNION ALL SELECT 1 FROM " + + FilesetMetaMapper.META_TABLE_NAME + + " WHERE schema_id = #{schemaId} AND deleted_at = 0" + + " UNION ALL SELECT 1 FROM " + + FunctionMetaMapper.TABLE_NAME + + " WHERE schema_id = #{schemaId} AND deleted_at = 0" + + " UNION ALL SELECT 1 FROM " + + ModelMetaMapper.TABLE_NAME + + " WHERE schema_id = #{schemaId} AND deleted_at = 0" + + " UNION ALL SELECT 1 FROM " + + TopicMetaMapper.TABLE_NAME + + " WHERE schema_id = #{schemaId} AND deleted_at = 0" + + " LIMIT 1"; + } + /** Returns SQL that selects and locks an active schema by ID. */ public String selectSchemaMetaByIdForUpdate(@Param("schemaId") Long schemaId) { return selectSchemaMetaById(schemaId) + " FOR UPDATE"; diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/service/FilesetMetaService.java b/core/src/main/java/org/apache/gravitino/storage/relational/service/FilesetMetaService.java index 70ce93113aa..855c08a1d8e 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/service/FilesetMetaService.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/service/FilesetMetaService.java @@ -26,7 +26,6 @@ import java.util.List; import java.util.Map; import java.util.Objects; -import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Function; import java.util.stream.Collectors; @@ -171,56 +170,51 @@ public void insertFileset(FilesetEntity filesetEntity, boolean overwrite) throws // The schema lock, metadata row, and every storage-location version row share one // transaction. A failure in any later step restores all earlier writes. - 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( - filesetEntity.nameIdentifier(), - po.getSchemaId(), - po.getCatalogId(), - po.getMetalakeId()), - () -> - SessionUtils.doWithoutCommit( - FilesetMetaMapper.class, - mapper -> { - FilesetPO storedPO = - overwrite - ? mapper.selectFilesetMetaBySchemaIdAndNameForUpdate( - po.getSchemaId(), po.getFilesetName()) - : null; - if (storedPO == null) { - mapper.insertFilesetMeta(po); - return; - } - - // Resolve the natural-key overwrite before building its snapshot. This keeps - // the stored ID in both the metadata row and identifier property without a - // post-insert JSON/PO rewrite. - FilesetEntity replacement = - filesetWithPersistedId(filesetEntity, storedPO.getFilesetId()); - Long maxStoredVersion = - SessionUtils.getWithoutCommit( - FilesetVersionMapper.class, - versionMapper -> - versionMapper.selectMaxFilesetVersion(storedPO.getFilesetId())); - FilesetPO replacementPO = - POConverters.updateFilesetPOWithVersion( - storedPO, replacement, maxStoredVersion); - Integer updated = mapper.updateFilesetMeta(replacementPO, storedPO); - Preconditions.checkState( - updated != null && updated == 1, - "The overwritten fileset %s in schema %s changed while its row was held", - po.getFilesetName(), - po.getSchemaId()); - persistedPO.set(replacementPO); - }), - () -> - SessionUtils.doWithoutCommit( - FilesetVersionMapper.class, - mapper -> - mapper.insertFilesetVersions(persistedPO.get().getFilesetVersionPOs()))); + SchemaMetaService.getInstance() + .doWithSchemaWriteLock( + filesetEntity.nameIdentifier(), + po.getSchemaId(), + po.getCatalogId(), + po.getMetalakeId(), + () -> + SessionUtils.doWithoutCommit( + FilesetMetaMapper.class, + mapper -> { + FilesetPO storedPO = + overwrite + ? mapper.selectFilesetMetaBySchemaIdAndNameForUpdate( + po.getSchemaId(), po.getFilesetName()) + : null; + if (storedPO == null) { + mapper.insertFilesetMeta(po); + return; + } + + // Resolve the natural-key overwrite before building its snapshot so the + // metadata row and identifier property both keep the stored ID. + FilesetEntity replacement = + filesetWithPersistedId(filesetEntity, storedPO.getFilesetId()); + Long maxStoredVersion = + SessionUtils.getWithoutCommit( + FilesetVersionMapper.class, + versionMapper -> + versionMapper.selectMaxFilesetVersion(storedPO.getFilesetId())); + FilesetPO replacementPO = + POConverters.updateFilesetPOWithVersion( + storedPO, replacement, maxStoredVersion); + Integer updated = mapper.updateFilesetMeta(replacementPO, storedPO); + Preconditions.checkState( + updated != null && updated == 1, + "The overwritten fileset %s in schema %s changed while its row was held", + po.getFilesetName(), + po.getSchemaId()); + persistedPO.set(replacementPO); + }), + () -> + SessionUtils.doWithoutCommit( + FilesetVersionMapper.class, + mapper -> + mapper.insertFilesetVersions(persistedPO.get().getFilesetVersionPOs()))); } catch (RuntimeException re) { ExceptionUtils.checkSQLException( re, Entity.EntityType.FILESET, filesetEntity.nameIdentifier().toString()); @@ -244,27 +238,14 @@ public FilesetEntity updateFileset( oldFilesetEntity.id()); try { - FilesetPO newFilesetPO = - POConverters.updateFilesetPOWithVersion(oldFilesetPO, newEntity, null); - if (tryUpdateFileset(newFilesetPO, oldFilesetPO)) { - return newEntity; - } - - // The metadata CAS also rejects a version that already has an active stored snapshot. Only - // that uncommon legacy case needs the MAX(version) round trip; normal alters finish above. - Long maxStoredVersion = - SessionUtils.getWithoutCommit( - FilesetVersionMapper.class, - mapper -> mapper.selectMaxFilesetVersion(oldFilesetPO.getFilesetId())); - if (maxStoredVersion != null - && maxStoredVersion >= newFilesetPO.getCurrentVersion() - && tryUpdateFileset( - POConverters.updateFilesetPOWithVersion(oldFilesetPO, newEntity, maxStoredVersion), - oldFilesetPO)) { - return newEntity; - } - - throw filesetWriteFailure(identifier, oldFilesetPO); + SchemaMetaService.getInstance() + .doWithSchemaWriteLock( + identifier, + oldFilesetPO.getSchemaId(), + oldFilesetPO.getCatalogId(), + oldFilesetPO.getMetalakeId(), + () -> updateFilesetInTransaction(identifier, newEntity, oldFilesetPO)); + return newEntity; } catch (RuntimeException re) { ExceptionUtils.checkSQLException( re, Entity.EntityType.FILESET, newEntity.nameIdentifier().toString()); @@ -477,26 +458,44 @@ void deleteFilesetWithVersion(NameIdentifier identifier, FilesetPO observedFiles } } + private void updateFilesetInTransaction( + NameIdentifier identifier, FilesetEntity newEntity, FilesetPO oldFilesetPO) { + FilesetPO newFilesetPO = POConverters.updateFilesetPOWithVersion(oldFilesetPO, newEntity, null); + if (tryUpdateFileset(newFilesetPO, oldFilesetPO)) { + return; + } + + // The metadata CAS also rejects a version that already has an active stored snapshot. Only + // that uncommon legacy case needs the MAX(version) round trip; normal alters finish above. + Long maxStoredVersion = + SessionUtils.getWithoutCommit( + FilesetVersionMapper.class, + mapper -> mapper.selectMaxFilesetVersion(oldFilesetPO.getFilesetId())); + if (maxStoredVersion != null + && maxStoredVersion >= newFilesetPO.getCurrentVersion() + && tryUpdateFileset( + POConverters.updateFilesetPOWithVersion(oldFilesetPO, newEntity, maxStoredVersion), + oldFilesetPO)) { + return; + } + + throw filesetWriteFailure(identifier, oldFilesetPO); + } + private boolean tryUpdateFileset(FilesetPO newFilesetPO, FilesetPO oldFilesetPO) { - AtomicBoolean updated = new AtomicBoolean(false); - SessionUtils.doMultipleWithCommit( - () -> { - Integer updateCount = - SessionUtils.getWithoutCommit( - FilesetMetaMapper.class, - mapper -> mapper.updateFilesetMeta(newFilesetPO, oldFilesetPO)); - updated.set(updateCount != null && updateCount > 0); - }, - () -> { - if (updated.get()) { - // The metadata row now points to this complete snapshot. It stays in the same - // transaction so a failed version insert also restores the metadata version. - SessionUtils.doWithoutCommit( - FilesetVersionMapper.class, - mapper -> mapper.insertFilesetVersions(newFilesetPO.getFilesetVersionPOs())); - } - }); - return updated.get(); + Integer updateCount = + SessionUtils.getWithoutCommit( + FilesetMetaMapper.class, + mapper -> mapper.updateFilesetMeta(newFilesetPO, oldFilesetPO)); + boolean updated = updateCount != null && updateCount > 0; + if (updated) { + // The metadata row now points to this complete snapshot. The caller's schema transaction + // ensures a failed version insert also restores the metadata version. + SessionUtils.doWithoutCommit( + FilesetVersionMapper.class, + mapper -> mapper.insertFilesetVersions(newFilesetPO.getFilesetVersionPOs())); + } + return updated; } private FilesetEntity filesetWithPersistedId(FilesetEntity filesetEntity, Long persistedId) { diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/service/FunctionMetaService.java b/core/src/main/java/org/apache/gravitino/storage/relational/service/FunctionMetaService.java index 04976bed87a..4806e718be2 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/service/FunctionMetaService.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/service/FunctionMetaService.java @@ -114,29 +114,26 @@ public void insertFunction(FunctionEntity functionEntity, boolean overwrite) thr fillFunctionPOBuilderParentEntityId(builder, functionEntity.namespace()); FunctionPO po = initializeFunctionPO(functionEntity, builder); - SessionUtils.doMultipleWithCommit( - // Hold the parent schema row until this transaction ends, so the function cannot be - // written below a schema that is being dropped. - () -> - SchemaMetaService.getInstance() - .lockSchemaForEntityWrite( - functionEntity.nameIdentifier(), - po.schemaId(), - po.catalogId(), - po.metalakeId()), - () -> - SessionUtils.doWithoutCommit( - FunctionMetaMapper.class, mapper -> ops.insertPO(mapper, po, overwrite)), - () -> - SessionUtils.doWithoutCommit( - FunctionVersionMetaMapper.class, - mapper -> { - if (overwrite) { - mapper.insertFunctionVersionMetaOnDuplicateKeyUpdate(po.functionVersionPO()); - } else { - mapper.insertFunctionVersionMeta(po.functionVersionPO()); - } - })); + SchemaMetaService.getInstance() + .doWithSchemaWriteLock( + functionEntity.nameIdentifier(), + po.schemaId(), + po.catalogId(), + po.metalakeId(), + () -> + SessionUtils.doWithoutCommit( + FunctionMetaMapper.class, mapper -> ops.insertPO(mapper, po, overwrite)), + () -> + SessionUtils.doWithoutCommit( + FunctionVersionMetaMapper.class, + mapper -> { + if (overwrite) { + mapper.insertFunctionVersionMetaOnDuplicateKeyUpdate( + po.functionVersionPO()); + } else { + mapper.insertFunctionVersionMeta(po.functionVersionPO()); + } + })); } catch (RuntimeException re) { ExceptionUtils.checkSQLException( re, Entity.EntityType.FUNCTION, functionEntity.nameIdentifier().toString()); @@ -271,33 +268,29 @@ public FunctionEntity updateFunction( try { FunctionPO newFunctionPO = updateFunctionPO(oldFunctionPO, newEntity); // Insert a new version and update function meta - SessionUtils.doMultipleWithCommit( - // The function was read before this transaction started. Lock its observed parent again - // before writing, so a schema drop cannot finish its function cleanup and then let this - // update add a new version below the deleted schema. - () -> - SchemaMetaService.getInstance() - .lockSchemaForEntityWrite( - identifier, - oldFunctionPO.schemaId(), - oldFunctionPO.catalogId(), - oldFunctionPO.metalakeId()), - () -> - SessionUtils.doWithoutCommit( - FunctionVersionMetaMapper.class, - mapper -> mapper.insertFunctionVersionMeta(newFunctionPO.functionVersionPO())), - () -> { - int updated = - SessionUtils.getWithoutCommit( - FunctionMetaMapper.class, - mapper -> ops.updatePO(mapper, newFunctionPO, oldFunctionPO)); - if (updated == 0) { - // The version row was inserted earlier in this transaction. Throwing here rolls the - // whole transaction back instead of leaving that version without an active function - // metadata row. - throw ExceptionUtils.concurrentModification(Entity.EntityType.FUNCTION, identifier); - } - }); + SchemaMetaService.getInstance() + .doWithSchemaWriteLock( + identifier, + oldFunctionPO.schemaId(), + oldFunctionPO.catalogId(), + oldFunctionPO.metalakeId(), + () -> + SessionUtils.doWithoutCommit( + FunctionVersionMetaMapper.class, + mapper -> + mapper.insertFunctionVersionMeta(newFunctionPO.functionVersionPO())), + () -> { + int updated = + SessionUtils.getWithoutCommit( + FunctionMetaMapper.class, + mapper -> ops.updatePO(mapper, newFunctionPO, oldFunctionPO)); + if (updated == 0) { + // The version was inserted above. Throwing here rolls it back instead of leaving + // an active version without function metadata. + throw ExceptionUtils.concurrentModification( + Entity.EntityType.FUNCTION, identifier); + } + }); return newEntity; } catch (RuntimeException re) { diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/service/ModelMetaService.java b/core/src/main/java/org/apache/gravitino/storage/relational/service/ModelMetaService.java index 00ef4aecea3..49756e860c5 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/service/ModelMetaService.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/service/ModelMetaService.java @@ -98,26 +98,22 @@ public void insertModel(ModelEntity modelEntity, boolean overwrite) throws IOExc fillModelPOBuilderParentEntityId(builder, modelEntity.namespace()); ModelPO po = POConverters.initializeModelPO(modelEntity, builder); - SessionUtils.doMultipleWithCommit( - // Hold the parent schema row until this transaction ends, so the model cannot be - // written below a schema that is being dropped. - () -> - SchemaMetaService.getInstance() - .lockSchemaForEntityWrite( - modelEntity.nameIdentifier(), - po.getSchemaId(), - po.getCatalogId(), - po.getMetalakeId()), - () -> - SessionUtils.doWithoutCommit( - ModelMetaMapper.class, - mapper -> { - if (overwrite) { - mapper.insertModelMetaOnDuplicateKeyUpdate(po); - } else { - mapper.insertModelMeta(po); - } - })); + SchemaMetaService.getInstance() + .doWithSchemaWriteLock( + modelEntity.nameIdentifier(), + po.getSchemaId(), + po.getCatalogId(), + po.getMetalakeId(), + () -> + SessionUtils.doWithoutCommit( + ModelMetaMapper.class, + mapper -> { + if (overwrite) { + mapper.insertModelMetaOnDuplicateKeyUpdate(po); + } else { + mapper.insertModelMeta(po); + } + })); } catch (RuntimeException re) { ExceptionUtils.checkSQLException( re, Entity.EntityType.MODEL, modelEntity.nameIdentifier().toString()); @@ -393,26 +389,31 @@ public ModelEntity updateModel( AtomicInteger updateResult = new AtomicInteger(0); try { - SessionUtils.doMultipleWithCommit( - () -> - updateResult.set( - SessionUtils.getWithoutCommit( - ModelMetaMapper.class, + SchemaMetaService.getInstance() + .doWithSchemaWriteLock( + identifier, + oldModelPO.getSchemaId(), + oldModelPO.getCatalogId(), + oldModelPO.getMetalakeId(), + () -> + updateResult.set( + SessionUtils.getWithoutCommit( + ModelMetaMapper.class, + mapper -> + mapper.updateModelMeta( + POConverters.updateModelPO(oldModelPO, newEntity), oldModelPO))), + () -> { + if (isRenamed && updateResult.get() > 0) { + SessionUtils.doWithoutCommit( + EntityChangeLogMapper.class, mapper -> - mapper.updateModelMeta( - POConverters.updateModelPO(oldModelPO, newEntity), oldModelPO))), - () -> { - if (isRenamed && updateResult.get() > 0) { - SessionUtils.doWithoutCommit( - EntityChangeLogMapper.class, - mapper -> - mapper.insertEntityChange( - metalakeName, - Entity.EntityType.MODEL.name(), - oldFullName, - OperateType.ALTER)); - } - }); + mapper.insertEntityChange( + metalakeName, + Entity.EntityType.MODEL.name(), + oldFullName, + OperateType.ALTER)); + } + }); } catch (RuntimeException re) { ExceptionUtils.checkSQLException( re, Entity.EntityType.MODEL, newEntity.nameIdentifier().toString()); diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/service/ModelVersionMetaService.java b/core/src/main/java/org/apache/gravitino/storage/relational/service/ModelVersionMetaService.java index b9b90f5d942..57bf70e0394 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/service/ModelVersionMetaService.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/service/ModelVersionMetaService.java @@ -172,11 +172,9 @@ public void insertModelVersion(ModelVersionEntity modelVersionEntity) throws IOE POConverters.initializeModelVersionAliasRelPO(modelVersionEntity, modelId); try { - SessionUtils.doMultipleWithCommit( - // Model versions carry the schema ID directly, so they must take the same parent fence - // as models. Otherwise a schema cascade can pass its model-version cleanup and a - // concurrent registration can insert a new active version below the deleted schema. - () -> lockSchemaForModelVersionWrite(modelIdent, modelPO), + doWithSchemaWriteLock( + modelIdent, + modelPO, () -> SessionUtils.doWithoutCommit( ModelVersionMetaMapper.class, @@ -190,9 +188,8 @@ public void insertModelVersion(ModelVersionEntity modelVersionEntity) throws IOE mapper -> mapper.insertModelVersionAliasRels(aliasRelPOs)); }, () -> { - // If the model version is inserted successfully, update the model latest version. A - // zero result means the model disappeared after the read above, so the inserted version - // and aliases must roll back with this transaction. + // A missing model means the version and aliases inserted above must roll back with this + // transaction. int updated = SessionUtils.getWithoutCommit( ModelMetaMapper.class, mapper -> mapper.updateModelLatestVersion(modelId)); @@ -358,10 +355,9 @@ public ModelVersionEntity updateModelVersion( final AtomicInteger updateResult = new AtomicInteger(0); try { - SessionUtils.doMultipleWithCommit( - // URI and alias updates can reinsert active model-version rows, so they need the same - // schema fence as a new version registration. - () -> lockSchemaForModelVersionWrite(modelIdent, modelPO), + doWithSchemaWriteLock( + modelIdent, + modelPO, () -> { if (isModelVersionUriUpdated) { // delete old model version POs first @@ -446,14 +442,15 @@ private boolean isModelVersionUriUpdated( return !oldUris.equals(newUris); } - private void lockSchemaForModelVersionWrite( - NameIdentifier modelIdentifier, ModelPO observedModelPO) { + private void doWithSchemaWriteLock( + NameIdentifier modelIdentifier, ModelPO modelPO, Runnable... modelVersionWriteOperations) { SchemaMetaService.getInstance() - .lockSchemaForEntityWrite( + .doWithSchemaWriteLock( modelIdentifier, - observedModelPO.getSchemaId(), - observedModelPO.getCatalogId(), - observedModelPO.getMetalakeId()); + modelPO.getSchemaId(), + modelPO.getCatalogId(), + modelPO.getMetalakeId(), + modelVersionWriteOperations); } private NoSuchEntityException noSuchModelException(NameIdentifier modelIdentifier) { diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/service/SchemaMetaService.java b/core/src/main/java/org/apache/gravitino/storage/relational/service/SchemaMetaService.java index 40eacb0b715..0bddca710cc 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/service/SchemaMetaService.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/service/SchemaMetaService.java @@ -504,12 +504,30 @@ private void lockCatalogForSchemaDelete(NameIdentifier identifier, SchemaPO obse } /** - * 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. + * Runs schema-scoped writes while holding a shared lock on their parent schema. + * + *

This method owns the transaction boundary on purpose. If callers locked the schema in one + * transaction and wrote the child in another, the lock would be released too early and a schema + * deletion could slip between those two steps. Keeping the lock and every supplied operation in + * the same transaction makes that mistake impossible for callers of this entry point. */ - void lockSchemaForEntityWrite( + void doWithSchemaWriteLock( + NameIdentifier entityIdentifier, + Long observedSchemaId, + Long observedCatalogId, + Long observedMetalakeId, + Runnable... entityWriteOperations) { + Runnable[] transactionOperations = new Runnable[entityWriteOperations.length + 1]; + transactionOperations[0] = + () -> + lockSchemaForEntityWrite( + entityIdentifier, observedSchemaId, observedCatalogId, observedMetalakeId); + System.arraycopy( + entityWriteOperations, 0, transactionOperations, 1, entityWriteOperations.length); + SessionUtils.doMultipleWithCommit(transactionOperations); + } + + private void lockSchemaForEntityWrite( NameIdentifier entityIdentifier, Long observedSchemaId, Long observedCatalogId, @@ -585,49 +603,18 @@ private void deleteDescendantSchemasWithVersions( } } - /** - * 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. - */ + /** Checks that no active schema or metadata object is left below the schema. */ private void checkSchemaIsEmpty(NameIdentifier identifier, SchemaPO schemaPO) { boolean hasDescendantSchemas = !listDescendantSchemaPOs(schemaPO).isEmpty(); - boolean hasTables = - !SessionUtils.getWithoutCommit( - TableMetaMapper.class, - mapper -> mapper.listTablePOsBySchemaId(schemaPO.getSchemaId())) - .isEmpty(); - boolean hasFilesets = - !SessionUtils.getWithoutCommit( - FilesetMetaMapper.class, - mapper -> mapper.listFilesetPOsBySchemaId(schemaPO.getSchemaId())) - .isEmpty(); - boolean hasModels = - !SessionUtils.getWithoutCommit( - ModelMetaMapper.class, - mapper -> mapper.listModelPOsBySchemaId(schemaPO.getSchemaId())) - .isEmpty(); - boolean hasTopics = - !SessionUtils.getWithoutCommit( - TopicMetaMapper.class, - mapper -> mapper.listTopicPOsBySchemaId(schemaPO.getSchemaId())) - .isEmpty(); - boolean hasViews = - !SessionUtils.getWithoutCommit( - ViewMetaMapper.class, - mapper -> mapper.listViewPOsBySchemaId(schemaPO.getSchemaId())) - .isEmpty(); - boolean hasFunctions = - !SessionUtils.getWithoutCommit( - FunctionMetaMapper.class, - mapper -> mapper.listFunctionPOsBySchemaId(schemaPO.getSchemaId())) - .isEmpty(); - if (hasDescendantSchemas - || hasTables - || hasFilesets - || hasModels - || hasTopics - || hasViews - || hasFunctions) { + // A non-cascade delete only needs to know whether any direct child exists. Asking the database + // for one literal avoids building every child PO and loading its version details while the + // schema delete lock is held. + boolean hasDirectChild = + SessionUtils.getWithoutCommit( + SchemaMetaMapper.class, + mapper -> mapper.selectActiveChildBySchemaId(schemaPO.getSchemaId())) + != null; + if (hasDescendantSchemas || hasDirectChild) { throw new NonEmptyEntityException( "Entity %s has sub-entities, you should remove sub-entities first", identifier); } diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/service/TableMetaService.java b/core/src/main/java/org/apache/gravitino/storage/relational/service/TableMetaService.java index a387bc370fd..6402a241bac 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/service/TableMetaService.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/service/TableMetaService.java @@ -117,69 +117,61 @@ public void insertTable(TableEntity tableEntity, boolean overwrite) throws IOExc AtomicReference persistedPO = new AtomicReference<>(po); // The schema lock, table row, version row, and columns share one transaction. If any later // step fails, the earlier inserts are rolled back as well. - SessionUtils.doMultipleWithCommit( - // Hold the parent schema row until this transaction ends, so the table cannot be - // written below a schema that is being dropped. - () -> - SchemaMetaService.getInstance() - .lockSchemaForEntityWrite( - tableEntity.nameIdentifier(), - po.getSchemaId(), - po.getCatalogId(), - po.getMetalakeId()), - () -> - SessionUtils.doWithoutCommit( - TableMetaMapper.class, - mapper -> { - ops.insertPO(mapper, po, overwrite); - if (overwrite) { - // MySQL may resolve the upsert through the active (schema_id, table_name, - // deleted_at) key rather than table_id. In that case it preserves the - // winner's ID. The upsert already holds that row until commit, so read the - // database-derived identity and version back through the same natural key. - TablePO storedPO = - mapper.selectTableMetaBySchemaIdAndName( - po.getSchemaId(), po.getTableName()); - Preconditions.checkState( - storedPO != null, - "The overwritten table %s in schema %s does not exist", - po.getTableName(), - po.getSchemaId()); - persistedPO.set(tablePOWithPersistedIdentityAndVersions(po, storedPO)); - } - }), - () -> - SessionUtils.doWithoutCommit( - TableVersionMapper.class, - mapper -> { - if (overwrite) { - TablePO storedPO = persistedPO.get(); - // Retire the version row this overwrite replaces. There is one only when the - // upsert updated an existing table: the database then moved the version from - // N to N + 1, so the row to retire is N. When the upsert inserted a brand new - // table the version is still the initial one and no earlier row exists. - if (storedPO.getCurrentVersion() > POConverters.INIT_VERSION) { - mapper.softDeleteTableVersionByTableIdAndVersion( - storedPO.getTableId(), storedPO.getCurrentVersion() - 1); - } - mapper.insertTableVersionOnDuplicateKeyUpdate(storedPO); - } else { - mapper.insertTableVersion(po); - } - }), - () -> { - // We need to delete the columns first if we want to overwrite the table. - if (overwrite) { - TableColumnMetaService.getInstance() - .deleteColumnsByTableId(persistedPO.get().getTableId()); - } - }, - () -> { - if (tableEntity.columns() != null && !tableEntity.columns().isEmpty()) { - TableColumnMetaService.getInstance() - .insertColumnPOs(persistedPO.get(), tableEntity.columns()); - } - }); + SchemaMetaService.getInstance() + .doWithSchemaWriteLock( + tableEntity.nameIdentifier(), + po.getSchemaId(), + po.getCatalogId(), + po.getMetalakeId(), + () -> + SessionUtils.doWithoutCommit( + TableMetaMapper.class, + mapper -> { + ops.insertPO(mapper, po, overwrite); + if (overwrite) { + // MySQL may preserve the existing table ID during an upsert. Read the + // stored identity and database-generated version while the row is locked. + TablePO storedPO = + mapper.selectTableMetaBySchemaIdAndName( + po.getSchemaId(), po.getTableName()); + Preconditions.checkState( + storedPO != null, + "The overwritten table %s in schema %s does not exist", + po.getTableName(), + po.getSchemaId()); + persistedPO.set(tablePOWithPersistedIdentityAndVersions(po, storedPO)); + } + }), + () -> + SessionUtils.doWithoutCommit( + TableVersionMapper.class, + mapper -> { + if (overwrite) { + TablePO storedPO = persistedPO.get(); + // An existing table advances from N to N + 1 during the upsert, so retire + // N before recording the new current version. A new table has no N row. + if (storedPO.getCurrentVersion() > POConverters.INIT_VERSION) { + mapper.softDeleteTableVersionByTableIdAndVersion( + storedPO.getTableId(), storedPO.getCurrentVersion() - 1); + } + mapper.insertTableVersionOnDuplicateKeyUpdate(storedPO); + } else { + mapper.insertTableVersion(po); + } + }), + () -> { + // We need to delete the columns first if we want to overwrite the table. + if (overwrite) { + TableColumnMetaService.getInstance() + .deleteColumnsByTableId(persistedPO.get().getTableId()); + } + }, + () -> { + if (tableEntity.columns() != null && !tableEntity.columns().isEmpty()) { + TableColumnMetaService.getInstance() + .insertColumnPOs(persistedPO.get(), tableEntity.columns()); + } + }); } catch (RuntimeException re) { ExceptionUtils.checkSQLException( @@ -216,57 +208,41 @@ public TableEntity updateTable( POConverters.updateTablePOWithVersionAndSchemaId(oldTablePO, newTableEntity, newSchemaId); try { - SessionUtils.doMultipleWithCommit( - () -> { - // Only an update that moves the table to another schema needs a lock here. The new - // parent must stay alive until the move commits; locking the old parent would not - // protect the table's new location. - if (isSchemaChanged) { - SchemaMetaService.getInstance() - .lockSchemaForEntityWrite( - newTableEntity.nameIdentifier(), - newSchemaId, - oldTablePO.getCatalogId(), - oldTablePO.getMetalakeId()); - } - }, - () -> { - // This update is the decision point for the whole transaction. current_version is the - // table's OCC token: if another writer changed the table after we read it, that writer - // has already increased the token and this UPDATE changes zero rows. Throwing here - // rolls back the transaction before it can touch the version history or columns. - int updated = - SessionUtils.getWithoutCommit( - TableMetaMapper.class, mapper -> ops.updatePO(mapper, newTablePO, oldTablePO)); - if (updated == 0) { - throw tableWriteFailure(identifier, oldTablePO); - } - }, - () -> { - // The table details live in table_version_info, keyed by (table_id, version), while - // table_meta only points at the current version. The two rows have to move together, - // and the upsert below has no version guard of its own: it overwrites whatever sits - // under that key. - // - // Say two writers both read version 5 and both want to write 6. Their version rows - // carry the same key, (table_id, 6), so whichever runs this statement second would - // silently replace the other's details. Ordering this step after the table_meta CAS is - // what prevents that: the loser matches no row up there, throws, and the transaction - // rolls back before reaching this statement. Only the winner ever writes version 6. - SessionUtils.doWithoutCommit( - TableVersionMapper.class, - mapper -> { - mapper.softDeleteTableVersionByTableIdAndVersion( - oldTablePO.getTableId(), oldTablePO.getCurrentVersion()); - mapper.insertTableVersionOnDuplicateKeyUpdate(newTablePO); - }); - }, - () -> { - // Column changes use the same new table version. Keeping this in the same transaction - // means a column failure also rolls back table_meta and table_version_info. - TableColumnMetaService.getInstance() - .updateColumnPOsFromTableDiff(oldTableEntity, newTableEntity, newTablePO); - }); + // For a cross-schema rename, the new schema is the parent that must remain alive. For a + // regular update, newSchemaId is the existing parent, so the same entry point covers both. + SchemaMetaService.getInstance() + .doWithSchemaWriteLock( + newTableEntity.nameIdentifier(), + newSchemaId, + oldTablePO.getCatalogId(), + oldTablePO.getMetalakeId(), + () -> { + // current_version is the table's OCC token. A zero-row update means another + // writer won, so stop before touching version history or columns. + int updated = + SessionUtils.getWithoutCommit( + TableMetaMapper.class, + mapper -> ops.updatePO(mapper, newTablePO, oldTablePO)); + if (updated == 0) { + throw tableWriteFailure(identifier, oldTablePO); + } + }, + () -> + SessionUtils.doWithoutCommit( + TableVersionMapper.class, + mapper -> { + // Only the CAS winner can reach this step, so it is safe to replace the + // details stored under the next table version. + mapper.softDeleteTableVersionByTableIdAndVersion( + oldTablePO.getTableId(), oldTablePO.getCurrentVersion()); + mapper.insertTableVersionOnDuplicateKeyUpdate(newTablePO); + }), + () -> { + // A column failure rolls back the table row and version row in the same + // transaction. + TableColumnMetaService.getInstance() + .updateColumnPOsFromTableDiff(oldTableEntity, newTableEntity, newTablePO); + }); } catch (RuntimeException re) { ExceptionUtils.checkSQLException( diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/service/TopicMetaService.java b/core/src/main/java/org/apache/gravitino/storage/relational/service/TopicMetaService.java index ca33b4fe2e7..bd4761c2c0d 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/service/TopicMetaService.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/service/TopicMetaService.java @@ -72,26 +72,22 @@ public void insertTopic(TopicEntity topicEntity, boolean overwrite) throws IOExc fillTopicPOBuilderParentEntityId(builder, topicEntity.namespace()); TopicPO po = POConverters.initializeTopicPOWithVersion(topicEntity, builder); - SessionUtils.doMultipleWithCommit( - // Hold the parent schema row until this transaction ends, so the topic cannot be - // written below a schema that is being dropped. - () -> - SchemaMetaService.getInstance() - .lockSchemaForEntityWrite( - topicEntity.nameIdentifier(), - po.getSchemaId(), - po.getCatalogId(), - po.getMetalakeId()), - () -> - SessionUtils.doWithoutCommit( - TopicMetaMapper.class, - mapper -> { - if (overwrite) { - mapper.insertTopicMetaOnDuplicateKeyUpdate(po); - } else { - mapper.insertTopicMeta(po); - } - })); + SchemaMetaService.getInstance() + .doWithSchemaWriteLock( + topicEntity.nameIdentifier(), + po.getSchemaId(), + po.getCatalogId(), + po.getMetalakeId(), + () -> + SessionUtils.doWithoutCommit( + TopicMetaMapper.class, + mapper -> { + if (overwrite) { + mapper.insertTopicMetaOnDuplicateKeyUpdate(po); + } else { + mapper.insertTopicMeta(po); + } + })); // TODO: insert topic dataLayout version after supporting it } catch (RuntimeException re) { ExceptionUtils.checkSQLException( @@ -124,15 +120,20 @@ public TopicEntity updateTopic( AtomicInteger updateResult = new AtomicInteger(0); try { - SessionUtils.doMultipleWithCommit( - () -> - updateResult.set( - SessionUtils.getWithoutCommit( - TopicMetaMapper.class, - mapper -> - mapper.updateTopicMeta( - POConverters.updateTopicPOWithVersion(oldTopicPO, newEntity), - oldTopicPO)))); + SchemaMetaService.getInstance() + .doWithSchemaWriteLock( + ident, + oldTopicPO.getSchemaId(), + oldTopicPO.getCatalogId(), + oldTopicPO.getMetalakeId(), + () -> + updateResult.set( + SessionUtils.getWithoutCommit( + TopicMetaMapper.class, + mapper -> + mapper.updateTopicMeta( + POConverters.updateTopicPOWithVersion(oldTopicPO, newEntity), + oldTopicPO)))); } catch (RuntimeException re) { ExceptionUtils.checkSQLException( re, Entity.EntityType.TOPIC, newEntity.nameIdentifier().toString()); diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/service/ViewMetaService.java b/core/src/main/java/org/apache/gravitino/storage/relational/service/ViewMetaService.java index 50ea6f72f07..0413d97320c 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/service/ViewMetaService.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/service/ViewMetaService.java @@ -105,29 +105,26 @@ public void insertView(ViewEntity viewEntity, boolean overwrite) throws IOExcept try { ViewPO po = initializeViewPO(viewEntity, builder); - SessionUtils.doMultipleWithCommit( - // Hold the parent schema row until this transaction ends, so the view cannot be - // written below a schema that is being dropped. - () -> - SchemaMetaService.getInstance() - .lockSchemaForEntityWrite( - viewEntity.nameIdentifier(), - po.getSchemaId(), - po.getCatalogId(), - po.getMetalakeId()), - () -> - SessionUtils.doWithoutCommit( - ViewMetaMapper.class, mapper -> ops.insertPO(mapper, po, overwrite)), - () -> - SessionUtils.doWithoutCommit( - ViewVersionInfoMapper.class, - mapper -> { - if (overwrite) { - mapper.insertViewVersionInfoOnDuplicateKeyUpdate(po.getViewVersionInfoPO()); - } else { - mapper.insertViewVersionInfo(po.getViewVersionInfoPO()); - } - })); + SchemaMetaService.getInstance() + .doWithSchemaWriteLock( + viewEntity.nameIdentifier(), + po.getSchemaId(), + po.getCatalogId(), + po.getMetalakeId(), + () -> + SessionUtils.doWithoutCommit( + ViewMetaMapper.class, mapper -> ops.insertPO(mapper, po, overwrite)), + () -> + SessionUtils.doWithoutCommit( + ViewVersionInfoMapper.class, + mapper -> { + if (overwrite) { + mapper.insertViewVersionInfoOnDuplicateKeyUpdate( + po.getViewVersionInfoPO()); + } else { + mapper.insertViewVersionInfo(po.getViewVersionInfoPO()); + } + })); } catch (RuntimeException re) { ExceptionUtils.checkSQLException( re, Entity.EntityType.VIEW, viewEntity.nameIdentifier().toString()); @@ -152,21 +149,32 @@ public ViewEntity updateView( AtomicInteger updateResult = new AtomicInteger(0); try { ViewPO newViewPO = updateViewPO(oldViewPO, newEntity); - SessionUtils.doMultipleWithCommit( - () -> - SessionUtils.doWithoutCommit( - ViewVersionInfoMapper.class, - mapper -> mapper.insertViewVersionInfo(newViewPO.getViewVersionInfoPO())), - () -> { - updateResult.set( - SessionUtils.getWithoutCommit( - ViewMetaMapper.class, mapper -> ops.updatePO(mapper, newViewPO, oldViewPO))); - if (updateResult.get() == 0) { - throw new RuntimeException("Failed to update the entity: " + ident); - } - }); + SchemaMetaService.getInstance() + .doWithSchemaWriteLock( + ident, + oldViewPO.getSchemaId(), + oldViewPO.getCatalogId(), + oldViewPO.getMetalakeId(), + () -> + SessionUtils.doWithoutCommit( + ViewVersionInfoMapper.class, + mapper -> mapper.insertViewVersionInfo(newViewPO.getViewVersionInfoPO())), + () -> { + updateResult.set( + SessionUtils.getWithoutCommit( + ViewMetaMapper.class, + mapper -> ops.updatePO(mapper, newViewPO, oldViewPO))); + if (updateResult.get() == 0) { + throw new RuntimeException("Failed to update the entity: " + ident); + } + }); return newEntity; } catch (RuntimeException re) { + // A missing parent is detected before the view update runs, so updateResult is still zero. + // Preserve that precise error instead of misreporting it as a view write conflict. + if (re instanceof NoSuchEntityException) { + throw re; + } if (updateResult.get() == 0) { throw new IOException("Failed to update the entity: " + ident); } diff --git a/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/base/TestSchemaMetaBaseSQLProvider.java b/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/base/TestSchemaMetaBaseSQLProvider.java new file mode 100644 index 00000000000..6cedd927d8b --- /dev/null +++ b/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/base/TestSchemaMetaBaseSQLProvider.java @@ -0,0 +1,61 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.gravitino.storage.relational.mapper.provider.base; + +import java.util.Arrays; +import java.util.List; +import org.apache.gravitino.storage.relational.mapper.FilesetMetaMapper; +import org.apache.gravitino.storage.relational.mapper.FunctionMetaMapper; +import org.apache.gravitino.storage.relational.mapper.ModelMetaMapper; +import org.apache.gravitino.storage.relational.mapper.TableMetaMapper; +import org.apache.gravitino.storage.relational.mapper.TopicMetaMapper; +import org.apache.gravitino.storage.relational.mapper.ViewMetaMapper; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class TestSchemaMetaBaseSQLProvider { + + private static final SchemaMetaBaseSQLProvider PROVIDER = new SchemaMetaBaseSQLProvider(); + + @Test + void testSelectActiveChildChecksEverySupportedChildType() { + String sql = PROVIDER.selectActiveChildBySchemaId(null); + List childTables = + Arrays.asList( + TableMetaMapper.TABLE_NAME, + ViewMetaMapper.TABLE_NAME, + FilesetMetaMapper.META_TABLE_NAME, + FunctionMetaMapper.TABLE_NAME, + ModelMetaMapper.TABLE_NAME, + TopicMetaMapper.TABLE_NAME); + + childTables.forEach( + tableName -> + Assertions.assertTrue( + sql.contains( + "FROM " + tableName + " WHERE schema_id = #{schemaId} AND deleted_at = 0"), + () -> "Missing active-child check for " + tableName + " in: " + sql)); + Assertions.assertEquals(childTables.size() - 1, countOccurrences(sql, "UNION ALL")); + Assertions.assertTrue(sql.endsWith("LIMIT 1")); + } + + private static int countOccurrences(String value, String target) { + return (value.length() - value.replace(target, "").length()) / target.length(); + } +} diff --git a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestSchemaMetaService.java b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestSchemaMetaService.java index 72d721a4c28..30b50efdbbb 100644 --- a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestSchemaMetaService.java +++ b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestSchemaMetaService.java @@ -31,6 +31,7 @@ import java.util.Arrays; import java.util.Collections; import java.util.List; +import java.util.Locale; import java.util.Objects; import java.util.Set; import java.util.concurrent.CountDownLatch; @@ -140,49 +141,7 @@ public void testSchemaChildServicesWaitForConcurrentSchemaDelete() throws Except createAndInsertMakeLake(metalakeName); createAndInsertCatalog(metalakeName, catalogName); - List childWrites = - Arrays.asList( - namespace -> - backend.insert( - createTableEntity( - RandomIdGenerator.INSTANCE.nextId(), namespace, "child_table", AUDIT_INFO), - false), - namespace -> - backend.insert( - createViewEntity(RandomIdGenerator.INSTANCE.nextId(), namespace, "child_view"), - false), - namespace -> - backend.insert( - createFilesetEntity( - RandomIdGenerator.INSTANCE.nextId(), - namespace, - "child_fileset", - AUDIT_INFO), - false), - namespace -> - backend.insert( - createFunctionEntity( - RandomIdGenerator.INSTANCE.nextId(), - namespace, - "child_function", - AUDIT_INFO), - false), - namespace -> - backend.insert( - createModelEntity( - RandomIdGenerator.INSTANCE.nextId(), - namespace, - "child_model", - "model comment", - 0, - Collections.emptyMap(), - AUDIT_INFO), - false), - namespace -> - backend.insert( - createTopicEntity( - RandomIdGenerator.INSTANCE.nextId(), namespace, "child_topic", AUDIT_INFO), - false)); + List childWrites = schemaChildWrites(); for (int index = 0; index < childWrites.size(); index++) { SchemaEntity schema = @@ -192,19 +151,76 @@ public void testSchemaChildServicesWaitForConcurrentSchemaDelete() throws Except "schema_for_entity_lock_" + index, AUDIT_INFO); backend.insert(schema, false); - assertChildWriteWaitsForConcurrentSchemaDelete(schema, childWrites.get(index)); + Namespace childNamespace = Namespace.of(metalakeName, catalogName, schema.name()); + SchemaChildWrite childWrite = childWrites.get(index); + assertSchemaChildActionWaitsForConcurrentDelete(schema, () -> childWrite.run(childNamespace)); } } - private void assertChildWriteWaitsForConcurrentSchemaDelete( - SchemaEntity schema, SchemaChildWrite childWrite) throws Exception { + @TestTemplate + public void testSchemaChildUpdatesWaitForConcurrentSchemaDelete() throws Exception { + createAndInsertMakeLake(metalakeName); + createAndInsertCatalog(metalakeName, catalogName); + + List childWrites = schemaChildWrites(); + List childTypes = schemaChildTypes(); + for (int index = 0; index < childWrites.size(); index++) { + SchemaEntity schema = + createSchemaEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofSchema(metalakeName, catalogName), + "schema_for_entity_update_lock_" + index, + AUDIT_INFO); + backend.insert(schema, false); + Namespace childNamespace = Namespace.of(metalakeName, catalogName, schema.name()); + childWrites.get(index).run(childNamespace); + + Entity.EntityType childType = childTypes.get(index); + NameIdentifier childIdentifier = + NameIdentifier.of(childNamespace, "child_" + childType.name().toLowerCase(Locale.ROOT)); + assertSchemaChildActionWaitsForConcurrentDelete( + schema, () -> backend.update(childIdentifier, childType, entity -> entity)); + } + } + + @TestTemplate + public void testSchemaActiveChildExistenceQueryCoversEveryChildType() throws Exception { + createAndInsertMakeLake(metalakeName); + createAndInsertCatalog(metalakeName, catalogName); + + List childWrites = schemaChildWrites(); + for (int index = 0; index < childWrites.size(); index++) { + SchemaEntity schema = + createSchemaEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofSchema(metalakeName, catalogName), + "schema_for_child_exists_" + index, + AUDIT_INFO); + backend.insert(schema, false); + + Assertions.assertNull(selectActiveSchemaChild(schema.id())); + childWrites.get(index).run(Namespace.of(metalakeName, catalogName, schema.name())); + Assertions.assertEquals(1, selectActiveSchemaChild(schema.id())); + + // Each UNION branch must protect the public non-cascade delete path, not merely return a + // value when the mapper is called directly. + assertThrows( + NonEmptyEntityException.class, + () -> SchemaMetaService.getInstance().deleteSchema(schema.nameIdentifier(), false)); + SchemaMetaService.getInstance().deleteSchema(schema.nameIdentifier(), true); + Assertions.assertNull(selectActiveSchemaChild(schema.id())); + } + } + + private void assertSchemaChildActionWaitsForConcurrentDelete( + SchemaEntity schema, SchemaChildAction childAction) throws Exception { SchemaPO observedSchemaPO = SessionUtils.getWithoutCommit( SchemaMetaMapper.class, mapper -> mapper.selectSchemaMetaById(schema.id())); CountDownLatch schemaDeleteLocked = new CountDownLatch(1); CountDownLatch allowDeleteCommit = new CountDownLatch(1); - CountDownLatch entityCreateStarted = new CountDownLatch(1); + CountDownLatch childActionStarted = new CountDownLatch(1); ExecutorService executor = Executors.newFixedThreadPool(2); Future deleteResult = executor.submit( @@ -235,26 +251,26 @@ private void assertChildWriteWaitsForConcurrentSchemaDelete( }); try { assertTrue(schemaDeleteLocked.await(30, TimeUnit.SECONDS)); - Future createResult = + Future childActionResult = executor.submit( () -> { - entityCreateStarted.countDown(); + childActionStarted.countDown(); try { // Exercise the real JDBCBackend-to-service path. This test must fail if any // schema-scoped service forgets to take the parent lock in its own transaction. - childWrite.run(Namespace.of(metalakeName, catalogName, schema.name())); + childAction.run(); return null; } catch (Throwable throwable) { return throwable; } }); - assertTrue(entityCreateStarted.await(30, TimeUnit.SECONDS)); - assertThrows(TimeoutException.class, () -> createResult.get(500, TimeUnit.MILLISECONDS)); + assertTrue(childActionStarted.await(30, TimeUnit.SECONDS)); + assertThrows(TimeoutException.class, () -> childActionResult.get(500, TimeUnit.MILLISECONDS)); allowDeleteCommit.countDown(); Assertions.assertNull(deleteResult.get(30, TimeUnit.SECONDS)); Assertions.assertInstanceOf( - NoSuchEntityException.class, createResult.get(30, TimeUnit.SECONDS)); + NoSuchEntityException.class, childActionResult.get(30, TimeUnit.SECONDS)); } finally { allowDeleteCommit.countDown(); executor.shutdownNow(); @@ -1210,6 +1226,65 @@ private int countActiveTagRelForMetadataObject(Long metadataObjectId, String met } } + private Integer selectActiveSchemaChild(Long schemaId) { + return SessionUtils.getWithoutCommit( + SchemaMetaMapper.class, mapper -> mapper.selectActiveChildBySchemaId(schemaId)); + } + + private List schemaChildWrites() { + return Arrays.asList( + namespace -> + backend.insert( + createTableEntity( + RandomIdGenerator.INSTANCE.nextId(), namespace, "child_table", AUDIT_INFO), + false), + namespace -> + backend.insert( + createViewEntity(RandomIdGenerator.INSTANCE.nextId(), namespace, "child_view"), + false), + namespace -> + backend.insert( + createFilesetEntity( + RandomIdGenerator.INSTANCE.nextId(), namespace, "child_fileset", AUDIT_INFO), + false), + namespace -> + backend.insert( + createFunctionEntity( + RandomIdGenerator.INSTANCE.nextId(), namespace, "child_function", AUDIT_INFO), + false), + namespace -> + backend.insert( + createModelEntity( + RandomIdGenerator.INSTANCE.nextId(), + namespace, + "child_model", + "model comment", + 0, + Collections.emptyMap(), + AUDIT_INFO), + false), + namespace -> + backend.insert( + createTopicEntity( + RandomIdGenerator.INSTANCE.nextId(), namespace, "child_topic", AUDIT_INFO), + false)); + } + + private List schemaChildTypes() { + return Arrays.asList( + Entity.EntityType.TABLE, + Entity.EntityType.VIEW, + Entity.EntityType.FILESET, + Entity.EntityType.FUNCTION, + Entity.EntityType.MODEL, + Entity.EntityType.TOPIC); + } + + @FunctionalInterface + private interface SchemaChildAction { + void run() throws Exception; + } + @FunctionalInterface private interface SchemaChildWrite { void run(Namespace namespace) throws Exception;