yuqi1129 commented on code in PR #13094:
URL: https://github.com/apache/gravitino/pull/13094#discussion_r4014025857
##########
core/src/test/java/org/apache/gravitino/storage/relational/service/TestSchemaMetaService.java:
##########
@@ -766,6 +766,366 @@ public void
testDeleteHierarchicalSchemaCascadeRemovesDescendantsAndChildren()
NameIdentifier.of(metalakeName, catalogName, "anc_a"),
Entity.EntityType.SCHEMA));
}
+ @TestTemplate
+ public void testSchemaChildUpdateServicesWaitForConcurrentSchemaDelete()
throws Exception {
+ createAndInsertMakeLake(metalakeName);
+ createAndInsertCatalog(metalakeName, catalogName);
+
+ List<SchemaChildUpdate> childUpdates =
+ Arrays.asList(
+ childIdent ->
+ TableMetaService.getInstance()
+ .updateTable(
+ childIdent,
+ entity -> {
+ TableEntity table = (TableEntity) entity;
+ return TableEntity.builder()
+ .withId(table.id())
+ .withName(table.name())
+ .withNamespace(table.namespace())
+ .withAuditInfo(table.auditInfo())
+ .withColumns(table.columns())
+ .withComment("updated table comment")
+ .withProperties(table.properties())
+ .build();
+ }),
+ childIdent ->
+ ViewMetaService.getInstance()
+ .updateView(
+ childIdent,
+ entity -> {
+ ViewEntity view = (ViewEntity) entity;
+ return ViewEntity.builder()
+ .withId(view.id())
+ .withName(view.name())
+ .withNamespace(view.namespace())
+ .withAuditInfo(view.auditInfo())
+ .withColumns(view.columns())
+ .withRepresentations(view.representations())
+ .withComment("updated view comment")
+ .build();
+ }),
+ childIdent ->
+ FilesetMetaService.getInstance()
+ .updateFileset(
+ childIdent,
+ entity -> {
+ FilesetEntity fileset = (FilesetEntity) entity;
+ return FilesetEntity.builder()
+ .withId(fileset.id())
+ .withName(fileset.name())
+ .withNamespace(fileset.namespace())
+ .withFilesetType(fileset.filesetType())
+ .withStorageLocations(fileset.storageLocations())
+ .withAuditInfo(fileset.auditInfo())
+ .withComment("updated fileset comment")
+ .withProperties(fileset.properties())
+ .build();
+ }),
+ childIdent ->
+ FunctionMetaService.getInstance()
+ .updateFunction(
+ childIdent,
+ entity -> {
+ FunctionEntity function = (FunctionEntity) entity;
+ return FunctionEntity.builder()
+ .withId(function.id())
+ .withName(function.name())
+ .withNamespace(function.namespace())
+ .withAuditInfo(function.auditInfo())
+ .withComment("updated function comment")
+ .withFunctionType(function.functionType())
+ .withDeterministic(function.deterministic())
+ .withDefinitions(function.definitions())
+ .build();
+ }),
+ childIdent ->
+ ModelMetaService.getInstance()
+ .updateModel(
+ childIdent,
+ entity -> {
+ ModelEntity model = (ModelEntity) entity;
+ return ModelEntity.builder()
+ .withId(model.id())
+ .withName(model.name())
+ .withNamespace(model.namespace())
+ .withAuditInfo(model.auditInfo())
+ .withComment("updated model comment")
+ .withLatestVersion(model.latestVersion())
+ .withProperties(model.properties())
+ .build();
+ }),
+ childIdent ->
+ TopicMetaService.getInstance()
+ .updateTopic(
+ childIdent,
+ entity -> {
+ TopicEntity topic = (TopicEntity) entity;
+ return TopicEntity.builder()
+ .withId(topic.id())
+ .withName(topic.name())
+ .withNamespace(topic.namespace())
+ .withAuditInfo(topic.auditInfo())
+ .withComment("updated topic comment")
+ .withProperties(topic.properties())
+ .build();
+ }));
+
+ for (int index = 0; index < childUpdates.size(); index++) {
+ String schemaName = "schema_for_update_lock_" + index;
+ SchemaEntity schema =
+ createSchemaEntity(
+ RandomIdGenerator.INSTANCE.nextId(),
+ NamespaceUtil.ofSchema(metalakeName, catalogName),
+ schemaName,
+ AUDIT_INFO);
+ backend.insert(schema, false);
+
+ Namespace schemaNamespace = Namespace.of(metalakeName, catalogName,
schemaName);
+
+ if (childUpdates.get(index) == childUpdates.get(0)) {
+ TableEntity table =
+ createTableEntity(
+ RandomIdGenerator.INSTANCE.nextId(), schemaNamespace,
"child_table", AUDIT_INFO);
+ backend.insert(table, false);
+ assertChildUpdateWaitsForConcurrentSchemaDelete(
+ schema, table.nameIdentifier(), childUpdates.get(index));
+ } else if (childUpdates.get(index) == childUpdates.get(1)) {
+ ViewEntity view =
+ createViewEntity(RandomIdGenerator.INSTANCE.nextId(),
schemaNamespace, "child_view");
+ backend.insert(view, false);
+ assertChildUpdateWaitsForConcurrentSchemaDelete(
+ schema, view.nameIdentifier(), childUpdates.get(index));
+ } else if (childUpdates.get(index) == childUpdates.get(2)) {
+ FilesetEntity fileset =
+ createFilesetEntity(
+ RandomIdGenerator.INSTANCE.nextId(), schemaNamespace,
"child_fileset", AUDIT_INFO);
+ backend.insert(fileset, false);
+ assertChildUpdateWaitsForConcurrentSchemaDelete(
+ schema, fileset.nameIdentifier(), childUpdates.get(index));
+ } else if (childUpdates.get(index) == childUpdates.get(3)) {
+ FunctionEntity function =
+ createFunctionEntity(
+ RandomIdGenerator.INSTANCE.nextId(), schemaNamespace,
"child_function", AUDIT_INFO);
+ backend.insert(function, false);
+ assertChildUpdateWaitsForConcurrentSchemaDelete(
+ schema, function.nameIdentifier(), childUpdates.get(index));
+ } else if (childUpdates.get(index) == childUpdates.get(4)) {
+ ModelEntity model =
+ createModelEntity(
+ RandomIdGenerator.INSTANCE.nextId(),
+ schemaNamespace,
+ "child_model",
+ "model comment",
+ 0,
+ Collections.emptyMap(),
+ AUDIT_INFO);
+ backend.insert(model, false);
+ assertChildUpdateWaitsForConcurrentSchemaDelete(
+ schema, model.nameIdentifier(), childUpdates.get(index));
+ } else if (childUpdates.get(index) == childUpdates.get(5)) {
+ TopicEntity topic =
+ createTopicEntity(
+ RandomIdGenerator.INSTANCE.nextId(), schemaNamespace,
"child_topic", AUDIT_INFO);
+ backend.insert(topic, false);
+ assertChildUpdateWaitsForConcurrentSchemaDelete(
+ schema, topic.nameIdentifier(), childUpdates.get(index));
+ }
+ }
+ }
+
+ private void assertChildUpdateWaitsForConcurrentSchemaDelete(
+ SchemaEntity schema, NameIdentifier childIdent, SchemaChildUpdate
childUpdate)
+ throws Exception {
+ // Run the schema cascade delete and child update concurrently.
+ // After both finish, assert that no orphan version rows remain.
+ // Only checks the no-orphan invariant; lock timing varies across backends.
+ CountDownLatch bothStarted = new CountDownLatch(2);
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+
+ Future<Throwable> deleteResult =
+ executor.submit(
+ () -> {
+ try {
+ bothStarted.countDown();
+ assertTrue(bothStarted.await(30, TimeUnit.SECONDS));
+
SchemaMetaService.getInstance().deleteSchema(schema.nameIdentifier(), true);
+ return null;
+ } catch (Throwable throwable) {
+ return throwable;
+ }
+ });
+
+ Future<Throwable> updateResult =
+ executor.submit(
+ () -> {
+ try {
+ bothStarted.countDown();
+ assertTrue(bothStarted.await(30, TimeUnit.SECONDS));
+ childUpdate.run(childIdent);
+ return null;
+ } catch (Throwable throwable) {
+ return throwable;
+ }
+ });
+
+ try {
+ Throwable deleteFailure = deleteResult.get(60, TimeUnit.SECONDS);
+ Throwable updateFailure = updateResult.get(60, TimeUnit.SECONDS);
+
+ // The cascade delete must succeed.
+ Assertions.assertNull(deleteFailure, () -> "Schema cascade delete
failed: " + deleteFailure);
+
+ // The update either succeeded (then cascade cleaned up) or failed
(schema gone).
+ // Both are acceptable as long as no orphan rows remain.
+ Assertions.assertTrue(
+ updateFailure == null
+ || updateFailure instanceof NoSuchEntityException
+ || updateFailure instanceof OptimisticLockException
+ || updateFailure instanceof IOException,
+ () -> "Unexpected update failure: " + updateFailure);
+
+ // Schema and child entity should both be gone.
+ Assertions.assertFalse(backend.exists(schema.nameIdentifier(),
Entity.EntityType.SCHEMA));
Review Comment:
[P2] This helper does not check the behavior its name and comments describe.
`bothStarted` only makes the two tasks start together. It does not ensure
that the update has read the old entity or acquired the schema lock before the
delete continues. The delete can finish before the update reads anything, or
the update can finish before the delete starts doing database work. Both cases
pass without testing the lock added by this PR.
There is also no check here for leftover child rows: the final assertion
only checks that the schema is gone. It does not check the child entity or any
version, column, or alias rows. Accepting every `IOException` can also hide a
database error, such as a deadlock reported by an update, as long as the delete
succeeds.
`testCascadeDeleteLeavesNoOrphanVersionRows` does check table-version rows,
but it has the same uncontrolled timing and does not cover the other entity
types.
Please make these tests control both execution orders:
- Update first: pause the real service update after it acquires the schema
lock, start the delete, check that the delete waits, then release the update
and check the final rows.
- Delete first: pause the update after it reads the old entity but before
its write transaction, complete the delete, then resume the update. Check that
it fails because the parent is gone and leaves no new rows.
For each entity type, check the metadata and its related tables directly by
entity ID, including rows with `deleted_at = 0`. Also assert the expected
failure type instead of accepting any `IOException`. These checks should be
part of this PR and run on MySQL and PostgreSQL as well as H2, since H2 uses an
exclusive lock in place of the shared schema lock.
##########
core/src/main/java/org/apache/gravitino/storage/relational/service/TableMetaService.java:
##########
@@ -218,14 +218,45 @@ public <E extends Entity & HasIdentifier> TableEntity
updateTable(
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.
+ // Always hold the parent schema row until the transaction ends,
so the
+ // table cannot be updated below a schema that is being dropped.
+ // For cross-schema moves, lock both source and destination
schemas.
if (isSchemaChanged) {
+ Long sourceSchemaId = oldTablePO.getSchemaId();
+ Long destSchemaId = newSchemaId;
+ // Lock the smaller schemaId first to avoid deadlocks
+ if (sourceSchemaId.compareTo(destSchemaId) <= 0) {
+ SchemaMetaService.getInstance()
+ .lockSchemaForEntityWrite(
+ oldTableEntity.nameIdentifier(),
+ sourceSchemaId,
+ oldTablePO.getCatalogId(),
+ oldTablePO.getMetalakeId());
+ SchemaMetaService.getInstance()
+ .lockSchemaForEntityWrite(
+ newTableEntity.nameIdentifier(),
+ destSchemaId,
+ oldTablePO.getCatalogId(),
+ oldTablePO.getMetalakeId());
Review Comment:
[P2] This lock order can deadlock with a hierarchical schema cascade delete.
Sorting the two schema IDs does not match the delete path.
`SchemaMetaService.deleteSchema` first locks the catalog, then updates the
target schema, and only then deletes its descendant schemas. Schema IDs are
random, so a child schema can have a smaller ID than its parent.
For example, let `a` have ID 200 and `a:b` have ID 100. A table move from
`a:b` to `a` can run as follows:
1. The move takes a shared lock on schema 100.
2. A concurrent cascade delete of `a` locks the catalog and updates schema
200, holding that row lock.
3. The move tries to lock schema 200 and waits for the delete.
4. The delete tries to delete descendant schema 100 and waits for the move.
Neither transaction can continue. I reproduced this SQL sequence on
PostgreSQL 14 and got `40P01: deadlock detected` in the cascade transaction.
PostgreSQL may abort either transaction, so this can fail either the delete or
the move. The same new locking code exists in
`FunctionMetaService.updateFunction` and `ViewMetaService.updateView`.
Please make the move and delete paths use a compatible lock order in this
PR. One limited change would be for cross-schema moves to take a shared lock on
the relevant catalog before taking either schema lock. The delete already takes
an exclusive catalog lock, so it would then wait before locking any schema. If
moves across catalogs are supported, take both catalog locks in a fixed order
before taking the schema locks. This adds a catalog lock lookup and makes
schema deletion wait for the move; same-schema updates can keep their current
approach.
Please also add a test with fixed IDs where the child schema ID is smaller
than the parent ID, and control the two threads at the lock steps above. Cover
the table, view, and function move paths.
--
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]
