LiJie20190102 commented on code in PR #13094:
URL: https://github.com/apache/gravitino/pull/13094#discussion_r4062257520
##########
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:
@yuqi1129 Can youreview it again
--
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]
