yuqi1129 commented on code in PR #13094:
URL: https://github.com/apache/gravitino/pull/13094#discussion_r4058996718
##########
core/src/main/java/org/apache/gravitino/storage/relational/service/SchemaMetaService.java:
##########
@@ -296,6 +297,10 @@ public boolean deleteSchema(NameIdentifier identifier,
boolean cascade) {
SessionUtils.doWithoutCommit(
TableMetaMapper.class,
mapper ->
mapper.softDeleteTableMetasBySchemaIds(schemaIds.get())),
+ () ->
+ SessionUtils.doWithoutCommit(
+ TableVersionMapper.class,
+ mapper ->
mapper.softDeleteTableVersionsBySchemaIds(schemaIds.get())),
Review Comment:
This closes the active `table_version_info` gap for schema cascade, but
catalog and metalake cascades still soft-delete `table_meta` without
soft-deleting its active version rows. For a table removed through either
parent cascade, those rows keep `deleted_at = 0` and are not eligible for
legacy-timeline cleanup. Could we apply the same table-ID-based cleanup to
those two cascade paths, or explicitly track that remaining scope if it belongs
in a follow-up?
##########
core/src/test/java/org/apache/gravitino/storage/relational/service/TestSchemaMetaService.java:
##########
@@ -777,6 +778,604 @@ public void
testDeleteHierarchicalSchemaCascadeRemovesDescendantsAndChildren()
NameIdentifier.of(metalakeName, catalogName, "anc_a"),
Entity.EntityType.SCHEMA));
}
+ @TestTemplate
+ public void testSchemaChildUpdateServicesWaitForConcurrentSchemaDelete()
throws Exception {
+ createAndInsertMakeLake(metalakeName);
+ createAndInsertCatalog(metalakeName, catalogName);
+
+ List<SchemaChildUpdateCase> childCases =
+ Arrays.asList(
+ new SchemaChildUpdateCase(
+ Entity.EntityType.TABLE,
+ "table_meta",
+ "table_id",
+ "table_version_info",
+ (namespace, name, bk) -> {
+ TableEntity e =
+ createTableEntity(
+ RandomIdGenerator.INSTANCE.nextId(), namespace,
name, AUDIT_INFO);
+ bk.insert(e, false);
+ return new Object[] {e.nameIdentifier(), e.id()};
+ },
+ 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();
+ })),
+ new SchemaChildUpdateCase(
+ Entity.EntityType.VIEW,
+ "view_meta",
+ "view_id",
+ "view_version_info",
+ (namespace, name, bk) -> {
+ ViewEntity e =
+ createViewEntity(RandomIdGenerator.INSTANCE.nextId(),
namespace, name);
+ bk.insert(e, false);
+ return new Object[] {e.nameIdentifier(), e.id()};
+ },
+ 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();
+ })),
+ new SchemaChildUpdateCase(
+ Entity.EntityType.FILESET,
+ "fileset_meta",
+ "fileset_id",
+ "fileset_version_info",
+ (namespace, name, bk) -> {
+ FilesetEntity e =
+ createFilesetEntity(
+ RandomIdGenerator.INSTANCE.nextId(), namespace,
name, AUDIT_INFO);
+ bk.insert(e, false);
+ return new Object[] {e.nameIdentifier(), e.id()};
+ },
+ 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();
+ })),
+ new SchemaChildUpdateCase(
+ Entity.EntityType.FUNCTION,
+ "function_meta",
+ "function_id",
+ "function_version_info",
+ (namespace, name, bk) -> {
+ FunctionEntity e =
+ createFunctionEntity(
+ RandomIdGenerator.INSTANCE.nextId(), namespace,
name, AUDIT_INFO);
+ bk.insert(e, false);
+ return new Object[] {e.nameIdentifier(), e.id()};
+ },
+ 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();
+ })),
+ new SchemaChildUpdateCase(
+ Entity.EntityType.MODEL,
+ "model_meta",
+ "model_id",
+ null,
+ (namespace, name, bk) -> {
+ ModelEntity e =
+ createModelEntity(
+ RandomIdGenerator.INSTANCE.nextId(),
+ namespace,
+ name,
+ "model comment",
+ 0,
+ Collections.emptyMap(),
+ AUDIT_INFO);
+ bk.insert(e, false);
+ return new Object[] {e.nameIdentifier(), e.id()};
+ },
+ 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();
+ })),
+ new SchemaChildUpdateCase(
+ Entity.EntityType.TOPIC,
+ "topic_meta",
+ "topic_id",
+ null,
+ (namespace, name, bk) -> {
+ TopicEntity e =
+ createTopicEntity(
+ RandomIdGenerator.INSTANCE.nextId(), namespace,
name, AUDIT_INFO);
+ bk.insert(e, false);
+ return new Object[] {e.nameIdentifier(), e.id()};
+ },
+ 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 < childCases.size(); index++) {
+ SchemaChildUpdateCase childCase = childCases.get(index);
+ String schemaName =
+ "schema_for_update_lock_" +
childCase.entityType.name().toLowerCase(Locale.ROOT);
+ SchemaEntity schema =
+ createSchemaEntity(
+ RandomIdGenerator.INSTANCE.nextId(),
+ NamespaceUtil.ofSchema(metalakeName, catalogName),
+ schemaName,
+ AUDIT_INFO);
+ backend.insert(schema, false);
+
+ Namespace schemaNamespace = Namespace.of(metalakeName, catalogName,
schemaName);
+ String childName = "child_" +
childCase.entityType.name().toLowerCase(Locale.ROOT);
+ Object[] ref = childCase.createChild.run(schemaNamespace, childName,
backend);
+ NameIdentifier childIdent = (NameIdentifier) ref[0];
+ Long childId = (Long) ref[1];
+
+ assertChildUpdateBlocksOnConcurrentSchemaDelete(
+ schema,
+ childIdent,
+ childCase.entityType,
+ childCase.update,
+ childId,
+ childCase.metaTable,
+ childCase.idColumn,
+ childCase.versionTable);
+ }
+ }
+
+ private void assertChildUpdateBlocksOnConcurrentSchemaDelete(
+ SchemaEntity schema,
+ NameIdentifier childIdent,
+ Entity.EntityType childType,
+ SchemaChildUpdate childUpdate,
+ Long childId,
+ String metaTable,
+ String idColumn,
+ String versionTable)
+ throws Exception {
+ SchemaPO observedSchemaPO =
+ SessionUtils.getWithoutCommit(
+ SchemaMetaMapper.class, mapper ->
mapper.selectSchemaMetaById(schema.id()));
+
+ CountDownLatch schemaDeleteLocked = new CountDownLatch(1);
+ CountDownLatch allowDeleteCommit = new CountDownLatch(1);
+ CountDownLatch updateStarted = new CountDownLatch(1);
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+
+ // Thread 1: soft-delete the schema row and hold the transaction open. The
+ // uncommitted UPDATE holds an exclusive row lock on the schema_meta row.
+ Future<Throwable> deleteResult =
+ executor.submit(
+ () -> {
+ try {
+ SessionUtils.doMultipleWithCommit(
+ () -> {
+ int deleted =
+ SessionUtils.getWithoutCommit(
+ SchemaMetaMapper.class,
+ mapper ->
+
mapper.softDeleteSchemaMetaBySchemaIdAndVersion(
+ observedSchemaPO.getSchemaId(),
+ observedSchemaPO.getCurrentVersion()));
+ Assertions.assertEquals(1, deleted);
+ schemaDeleteLocked.countDown();
+ try {
+ assertTrue(allowDeleteCommit.await(30,
TimeUnit.SECONDS));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ });
+ return null;
+ } catch (Throwable throwable) {
+ return throwable;
+ }
+ });
+
+ try {
+ assertTrue(schemaDeleteLocked.await(30, TimeUnit.SECONDS));
+
+ // Thread 2: attempt to update the child entity. The update's
doWithSchemaWriteLock
+ // must block on the schema row held by Thread 1's uncommitted
soft-delete. If the
+ // update completes within 500 ms, the schema lock was not acquired by
the update path.
+ Future<Throwable> updateResult =
+ executor.submit(
+ () -> {
+ updateStarted.countDown();
+ try {
+ childUpdate.run(childIdent);
+ return null;
+ } catch (Throwable throwable) {
+ return throwable;
+ }
+ });
+
+ assertTrue(updateStarted.await(30, TimeUnit.SECONDS));
+ assertThrows(TimeoutException.class, () -> updateResult.get(500,
TimeUnit.MILLISECONDS));
Review Comment:
The timeout does not prove the update reached the schema lock:
`updateStarted` is counted down before `childUpdate.run`, so a delayed worker
can make this pass without testing lock contention. This test also holds only a
manually soft-deleted schema row, while
`testCascadeDeleteLeavesNoOrphanVersionRows` completes the real cascade before
starting the update. Could we add controlled overlap with `deleteSchema(...,
true)` in both orders, then check active child/version rows? The new
cross-schema table/view/function move paths also need a fixed-ID parent/child
cascade case to guard the catalog-first lock order.
--
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]
