diff --git a/catalogs/catalog-fileset/src/main/java/org/apache/gravitino/catalog/fileset/FilesetCatalogOperations.java b/catalogs/catalog-fileset/src/main/java/org/apache/gravitino/catalog/fileset/FilesetCatalogOperations.java index 42fe482febe..ea5b32edde3 100644 --- a/catalogs/catalog-fileset/src/main/java/org/apache/gravitino/catalog/fileset/FilesetCatalogOperations.java +++ b/catalogs/catalog-fileset/src/main/java/org/apache/gravitino/catalog/fileset/FilesetCatalogOperations.java @@ -574,6 +574,11 @@ public Fileset createMultipleLocationFileset( try { store.put(filesetEntity, true /* overwrite */); + } catch (NoSuchEntityException exception) { + // The schema can disappear after the check near the start of this method. The relational + // store detects that race while taking the parent-schema lock; translate its storage-level + // exception into the catalog API's documented missing-schema exception. + throw new NoSuchSchemaException(exception, SCHEMA_DOES_NOT_EXIST_MSG, schemaIdent); } catch (IOException ioe) { throw new RuntimeException("Failed to create fileset " + ident, ioe); } diff --git a/catalogs/catalog-fileset/src/test/java/org/apache/gravitino/catalog/fileset/TestFilesetCatalogOperations.java b/catalogs/catalog-fileset/src/test/java/org/apache/gravitino/catalog/fileset/TestFilesetCatalogOperations.java index fe610e7f0f8..07416f2def1 100644 --- a/catalogs/catalog-fileset/src/test/java/org/apache/gravitino/catalog/fileset/TestFilesetCatalogOperations.java +++ b/catalogs/catalog-fileset/src/test/java/org/apache/gravitino/catalog/fileset/TestFilesetCatalogOperations.java @@ -57,6 +57,7 @@ import java.io.IOException; import java.net.ConnectException; import java.nio.file.Paths; +import java.time.Instant; import java.util.Arrays; import java.util.Collections; import java.util.HashMap; @@ -95,6 +96,7 @@ import org.apache.gravitino.connector.PropertyEntry; import org.apache.gravitino.credential.CredentialConstants; import org.apache.gravitino.exceptions.GravitinoRuntimeException; +import org.apache.gravitino.exceptions.NoSuchEntityException; import org.apache.gravitino.exceptions.NoSuchFilesetException; import org.apache.gravitino.exceptions.NoSuchSchemaException; import org.apache.gravitino.exceptions.NonEmptySchemaException; @@ -102,6 +104,11 @@ import org.apache.gravitino.file.FileInfo; import org.apache.gravitino.file.Fileset; import org.apache.gravitino.file.FilesetChange; +import org.apache.gravitino.meta.AuditInfo; +import org.apache.gravitino.meta.BaseMetalake; +import org.apache.gravitino.meta.CatalogEntity; +import org.apache.gravitino.meta.FilesetEntity; +import org.apache.gravitino.meta.SchemaVersion; import org.apache.gravitino.secret.SecretConstants; import org.apache.gravitino.secret.SecretManager; import org.apache.gravitino.secret.SecretMaterial; @@ -229,7 +236,7 @@ private static CatalogInfo randomCatalogInfo( } @BeforeAll - public static void setUp() throws IllegalAccessException { + public static void setUp() throws IOException, IllegalAccessException { Config config = Mockito.mock(Config.class); when(config.get(ENTITY_STORE)).thenReturn(RELATIONAL_ENTITY_STORE); when(config.get(ENTITY_RELATIONAL_STORE)).thenReturn(DEFAULT_ENTITY_RELATIONAL_STORE); @@ -269,6 +276,28 @@ public static void setUp() throws IllegalAccessException { store.initialize(config); idGenerator = new RandomIdGenerator(); + AuditInfo auditInfo = + AuditInfo.builder().withCreator("test").withCreateTime(Instant.now()).build(); + BaseMetalake metalake = + BaseMetalake.builder() + .withId(1L) + .withName("m1") + .withVersion(SchemaVersion.V_0_1) + .withAuditInfo(auditInfo) + .build(); + store.put(metalake, false); + + CatalogEntity catalog = + CatalogEntity.builder() + .withId(1L) + .withName("c1") + .withNamespace(Namespace.of("m1")) + .withProvider("fileset") + .withType(Catalog.Type.FILESET) + .withAuditInfo(auditInfo) + .build(); + store.put(catalog, false); + // Mock MetalakeMetaService metalakeMetaService = MetalakeMetaService.getInstance(); MetalakeMetaService spyMetaService = Mockito.spy(metalakeMetaService); @@ -424,14 +453,13 @@ public void testCreateSchemaWithNoLocation() throws IOException { final long testId = generateTestId(); final String name = "schema" + testId; final String comment = "comment" + testId; - Schema schema = createSchema(testId, name, comment, null, null); + Schema schema = createSchema(name, comment, null, null); Assertions.assertEquals(name, schema.name()); Assertions.assertEquals(comment, schema.comment()); Throwable exception = Assertions.assertThrows( - SchemaAlreadyExistsException.class, - () -> createSchema(testId, name, comment, null, null)); + SchemaAlreadyExistsException.class, () -> createSchema(name, comment, null, null)); Assertions.assertEquals( "Schema m1.c1.schema" + testId + " already exists", exception.getMessage()); } @@ -446,7 +474,7 @@ public void testCreateSchemaWithEmptyCatalogLocation() throws IOException { Throwable exception = Assertions.assertThrows( IllegalArgumentException.class, - () -> createSchema(testId, schemaName, comment, catalogPath, null)); + () -> createSchema(schemaName, comment, catalogPath, null)); Assertions.assertEquals( "The value of the catalog property " + FilesetCatalogPropertiesMetadata.LOCATION @@ -460,7 +488,7 @@ public void testCreateSchemaWithCatalogLocation() throws IOException { String name = "schema" + testId; final String comment = "comment" + testId; String catalogPath = TEST_ROOT_PATH + "/" + "catalog12"; - Schema schema = createSchema(testId, name, comment, catalogPath, null); + Schema schema = createSchema(name, comment, catalogPath, null); Assertions.assertEquals(name, schema.name()); Path schemaPath = new Path(catalogPath, name); @@ -472,7 +500,7 @@ public void testCreateSchemaWithCatalogLocation() throws IOException { // test placeholder in catalog location name = "schema" + testId + "_1"; catalogPath = TEST_ROOT_PATH + "/" + "{{catalog}}-{{schema}}"; - schema = createSchema(testId, name, comment, catalogPath, null); + schema = createSchema(name, comment, catalogPath, null); Assertions.assertEquals(name, schema.name()); schemaPath = new Path(catalogPath, name); @@ -482,7 +510,7 @@ public void testCreateSchemaWithCatalogLocation() throws IOException { // Test disable server-side FS operations. name = "schema" + testId + "_2"; catalogPath = TEST_ROOT_PATH + "/" + "catalog12_2"; - schema = createSchema(testId, name, comment, catalogPath, null, true); + schema = createSchema(name, comment, catalogPath, null, true); Assertions.assertEquals(name, schema.name()); // Schema path should not be existed if the server-side FS operations are disabled. @@ -497,7 +525,7 @@ public void testCreateSchemaWithSchemaLocation() throws IOException { final String comment = "comment" + testId; String catalogPath = TEST_ROOT_PATH + "/" + "catalog" + testId; String schemaPath = catalogPath + "/" + name; - Schema schema = createSchema(testId, name, comment, null, schemaPath); + Schema schema = createSchema(name, comment, null, schemaPath); Assertions.assertEquals(name, schema.name()); Path schemaPath1 = new Path(schemaPath); @@ -509,7 +537,7 @@ public void testCreateSchemaWithSchemaLocation() throws IOException { // test placeholder in schema location name = "schema" + testId + "_1"; schemaPath = catalogPath + "/" + "{{schema}}"; - schema = createSchema(testId, name, comment, null, schemaPath); + schema = createSchema(name, comment, null, schemaPath); Assertions.assertEquals(name, schema.name()); schemaPath1 = new Path(schemaPath); @@ -522,7 +550,7 @@ public void testCreateSchemaWithSchemaLocation() throws IOException { Throwable exception = Assertions.assertThrows( IllegalArgumentException.class, - () -> createSchema(testId, schemaName1, comment, null, schemaPath2)); + () -> createSchema(schemaName1, comment, null, schemaPath2)); Assertions.assertTrue( exception.getMessage().contains("Placeholder in location should not be empty"), exception.getMessage()); @@ -530,7 +558,7 @@ public void testCreateSchemaWithSchemaLocation() throws IOException { // Test disable server-side FS operations. name = "schema" + testId + "_3"; schemaPath = catalogPath + "/" + name; - schema = createSchema(testId, name, comment, null, schemaPath, true); + schema = createSchema(name, comment, null, schemaPath, true); Assertions.assertEquals(name, schema.name()); // Schema path should not be existed if the server-side FS operations are disabled. @@ -544,7 +572,7 @@ public void testCreateSchemaWithCatalogAndSchemaLocation() throws IOException { String comment = "comment" + testId; String catalogPath = TEST_ROOT_PATH + "/" + "catalog" + testId; String schemaPath = TEST_ROOT_PATH + "/" + "schema" + testId; - Schema schema = createSchema(testId, name, comment, catalogPath, schemaPath); + Schema schema = createSchema(name, comment, catalogPath, schemaPath); Assertions.assertEquals(name, schema.name()); Path schemaPath1 = new Path(schemaPath); @@ -560,7 +588,7 @@ public void testCreateSchemaWithCatalogAndSchemaLocation() throws IOException { name = "schema" + testId + "_1"; catalogPath = TEST_ROOT_PATH + "/" + "{{catalog}}"; schemaPath = TEST_ROOT_PATH + "/" + "{{schema}}"; - schema = createSchema(testId, name, comment, catalogPath, schemaPath); + schema = createSchema(name, comment, catalogPath, schemaPath); Assertions.assertEquals(name, schema.name()); schemaPath1 = new Path(schemaPath); @@ -573,7 +601,7 @@ public void testCreateSchemaWithCatalogAndSchemaLocation() throws IOException { name = "schema" + testId + "_2"; catalogPath = TEST_ROOT_PATH + "/" + "catalog14_2"; schemaPath = TEST_ROOT_PATH + "/" + "schema14_2"; - schema = createSchema(testId, name, comment, catalogPath, schemaPath, true); + schema = createSchema(name, comment, catalogPath, schemaPath, true); Assertions.assertEquals(name, schema.name()); // Schema path should not be existed if the server-side FS operations are disabled. @@ -587,7 +615,7 @@ public void testLoadSchema() throws IOException { String name = "schema" + testId; String comment = "comment" + testId; String catalogPath = TEST_ROOT_PATH + "/" + "catalog" + testId; - Schema schema = createSchema(testId, name, comment, catalogPath, null); + Schema schema = createSchema(name, comment, catalogPath, null); NameIdentifier otherSchema = NameIdentifierUtil.ofSchema("m1", "c1", "otherSchema"); Assertions.assertEquals(name, schema.name()); @@ -615,8 +643,8 @@ public void testListSchema() throws IOException { String comment1 = "comment" + testId1; String name2 = "schema" + testId2; String comment2 = "comment" + testId2; - createSchema(testId1, name1, comment1, null, null); - createSchema(testId2, name2, comment2, null, null); + createSchema(name1, comment1, null, null); + createSchema(name2, comment2, null, null); try (FilesetCatalogOperations ops = new FilesetCatalogOperations(store, secretManager)) { ops.initialize(Maps.newHashMap(), randomCatalogInfo(), FILESET_PROPERTIES_METADATA); @@ -634,7 +662,7 @@ public void testAlterSchema() throws IOException { String name = "schema" + testId; String comment = "comment" + testId; String catalogPath = TEST_ROOT_PATH + "/" + "catalog" + testId; - Schema schema = createSchema(testId, name, comment, catalogPath, null); + Schema schema = createSchema(name, comment, catalogPath, null); Assertions.assertEquals(name, schema.name()); try (FilesetCatalogOperations ops = new FilesetCatalogOperations(store, secretManager)) { @@ -682,7 +710,7 @@ public void testDropSchema() throws IOException { final String comment = "comment" + testId; final String catalogPath = TEST_ROOT_PATH + "/" + "catalog" + testId; - Schema schema = createSchema(testId, schemaName, comment, catalogPath, null); + Schema schema = createSchema(schemaName, comment, catalogPath, null); Assertions.assertEquals(schemaName, schema.name()); NameIdentifier id = NameIdentifierUtil.ofSchema("m1", "c1", schemaName); @@ -705,7 +733,7 @@ public void testDropSchema() throws IOException { Assertions.assertFalse(fs.exists(schemaPath)); // Test drop non-empty schema with cascade = false - createSchema(testId, schemaName, comment, catalogPath, null); + createSchema(schemaName, comment, catalogPath, null); Fileset fs1 = createFileset("fs1", schemaName, "comment", Fileset.Type.MANAGED, catalogPath, null); Path fs1Path = new Path(fs1.storageLocation()); @@ -721,7 +749,7 @@ public void testDropSchema() throws IOException { Assertions.assertFalse(fs.exists(fs1Path)); // Test drop both managed and external filesets - createSchema(testId, schemaName, comment, catalogPath, null); + createSchema(schemaName, comment, catalogPath, null); Fileset fs2 = createFileset("fs2", schemaName, "comment", Fileset.Type.MANAGED, catalogPath, null); Path fs2Path = new Path(fs2.storageLocation()); @@ -737,7 +765,7 @@ public void testDropSchema() throws IOException { Assertions.assertTrue(fs.exists(fs3Path)); // Test drop schema with different storage location - createSchema(testId, schemaName, comment, catalogPath, null); + createSchema(schemaName, comment, catalogPath, null); Path fs4Path = new Path(TEST_ROOT_PATH + "/fs4"); createFileset( "fs4", schemaName, "comment", Fileset.Type.MANAGED, catalogPath, fs4Path.toString()); @@ -754,7 +782,7 @@ public void testDropSchemaWithFSOpsDisabled() throws IOException { final String filesetName = "fileset" + testId; final String catalogPath = TEST_ROOT_PATH + "/" + "catalog" + testId; - Schema schema = createSchema(testId, schemaName, comment, catalogPath, null); + Schema schema = createSchema(schemaName, comment, catalogPath, null); Assertions.assertEquals(schemaName, schema.name()); NameIdentifier id = NameIdentifierUtil.ofSchema("m1", "c1", schemaName); @@ -770,7 +798,7 @@ public void testDropSchemaWithFSOpsDisabled() throws IOException { FileSystem fs = schemaPath.getFileSystem(new Configuration()); Assertions.assertTrue(fs.exists(schemaPath)); - createSchema(testId, schemaName, comment, catalogPath, null); + createSchema(schemaName, comment, catalogPath, null); Fileset fs1 = createFileset(filesetName, schemaName, comment, Fileset.Type.MANAGED, catalogPath, null); Path fs1Path = new Path(fs1.storageLocation()); @@ -803,7 +831,7 @@ public void testCreateLoadAndDeleteFilesetWithLocations( try (FilesetCatalogOperations ops = new FilesetCatalogOperations(store, secretManager)) { ops.initialize(catalogProps, randomCatalogInfo("m1", "c1"), FILESET_PROPERTIES_METADATA); if (!ops.schemaExists(schemaIdent)) { - createSchema(generateTestId(), schemaName, comment, catalogPath, schemaPath); + createSchema(schemaName, comment, catalogPath, schemaPath); } Fileset fileset = createFileset(name, schemaName, "comment", type, catalogPath, storageLocation); @@ -860,7 +888,7 @@ public void testCreateLoadAndDeleteFilesetWithLocationsWhenFSOpsDisabled( try (FilesetCatalogOperations ops = new FilesetCatalogOperations(store, secretManager)) { ops.initialize(catalogProps, randomCatalogInfo("m1", "c1"), FILESET_PROPERTIES_METADATA); if (!ops.schemaExists(schemaIdent)) { - createSchema(generateTestId(), schemaName, comment, catalogPath, schemaPath, true); + createSchema(schemaName, comment, catalogPath, schemaPath, true); } Fileset fileset; @@ -918,7 +946,7 @@ public void testCreateFilesetWithExceptions() throws IOException { final String comment = "comment" + testId; final String filesetName = "fileset" + testId; - createSchema(testId, schemaName, comment, null, null); + createSchema(schemaName, comment, null, null); NameIdentifier filesetIdent = NameIdentifier.of("m1", "c1", schemaName, filesetName); // If neither catalog location, nor schema location and storageLocation is specified. @@ -961,6 +989,46 @@ public void testCreateFilesetWithExceptions() throws IOException { } } + @Test + public void testCreateFilesetMapsSchemaDeletionDuringStoreWrite() throws IOException { + long testId = generateTestId(); + String schemaName = "schema" + testId; + String filesetName = "fileset" + testId; + String catalogPath = TEST_ROOT_PATH + "/catalog" + testId; + createSchema(schemaName, "comment", catalogPath, null, true); + + EntityStore racingStore = Mockito.spy(store); + NoSuchEntityException deletedSchema = + new NoSuchEntityException( + NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE, "schema", schemaName); + Mockito.doThrow(deletedSchema) + .when(racingStore) + .put(Mockito.any(FilesetEntity.class), Mockito.eq(true)); + + try (FilesetCatalogOperations ops = new FilesetCatalogOperations(racingStore, secretManager)) { + ops.initialize( + ImmutableMap.of(DISABLE_FILESYSTEM_OPS, "true", LOCATION, catalogPath), + randomCatalogInfo("m1", "c1"), + FILESET_PROPERTIES_METADATA); + NameIdentifier filesetIdent = NameIdentifier.of("m1", "c1", schemaName, filesetName); + Map filesetProperties = + StringIdentifier.newPropertiesWithId( + StringIdentifier.fromId(idGenerator.nextId()), Collections.emptyMap()); + + NoSuchSchemaException exception = + Assertions.assertThrows( + NoSuchSchemaException.class, + () -> + ops.createMultipleLocationFileset( + filesetIdent, + "comment", + Fileset.Type.MANAGED, + Collections.emptyMap(), + filesetProperties)); + Assertions.assertSame(deletedSchema, exception.getCause()); + } + } + @Test public void testListFilesets() throws IOException { final long testId = generateTestId(); @@ -968,7 +1036,7 @@ public void testListFilesets() throws IOException { String comment = "comment" + testId; String schemaPath = TEST_ROOT_PATH + "/" + schemaName; - createSchema(testId, schemaName, comment, null, schemaPath); + createSchema(schemaName, comment, null, schemaPath); String[] filesets = { "fileset" + testId + "_1", "fileset" + testId + "_2", "fileset" + testId + "_3" @@ -998,7 +1066,7 @@ public void testListFilesetFiles() throws IOException { final String schemaPath = TEST_ROOT_PATH + "/" + schemaName; final NameIdentifier filesetIdent = NameIdentifier.of("m1", "c1", schemaName, filesetName); - createSchema(testId, schemaName, comment, null, schemaPath); + createSchema(schemaName, comment, null, schemaPath); createFileset(filesetName, schemaName, comment, Fileset.Type.MANAGED, null, null); try (FilesetCatalogOperations ops = new FilesetCatalogOperations(store, secretManager)) { @@ -1048,7 +1116,7 @@ public void testListFilesetFilesWithFSOpsDisabled() throws Exception { final String schemaPath = TEST_ROOT_PATH + "/" + schemaName; final NameIdentifier filesetIdent = NameIdentifier.of("m1", "c1", schemaName, filesetName); - createSchema(testId, schemaName, comment, null, schemaPath); + createSchema(schemaName, comment, null, schemaPath); createFileset(filesetName, schemaName, comment, Fileset.Type.MANAGED, null, null); Map catalogProps = Collections.singletonMap(DISABLE_FILESYSTEM_OPS, "true"); @@ -1075,7 +1143,7 @@ public void testListFilesetFilesWithNonExistentPath() throws IOException { String filesetName = "fileset" + testId; final String nonExistentSubPath = "/non_existent_file.txt"; - Schema schema = createSchema(testId, schemaName, comment, null, schemaPath); + Schema schema = createSchema(schemaName, comment, null, schemaPath); Fileset fileset = createFileset(filesetName, schemaName, comment, Fileset.Type.MANAGED, null, null); final NameIdentifier filesetIdent = @@ -1117,7 +1185,7 @@ public void testRenameFileset( try (FilesetCatalogOperations ops = new FilesetCatalogOperations(store, secretManager)) { ops.initialize(catalogProps, randomCatalogInfo("m1", "c1"), FILESET_PROPERTIES_METADATA); if (!ops.schemaExists(schemaIdent)) { - createSchema(generateTestId(), schemaName, comment, catalogPath, schemaPath); + createSchema(schemaName, comment, catalogPath, schemaPath); } Fileset fileset = createFileset(name, schemaName, "comment", type, catalogPath, storageLocation); @@ -1155,7 +1223,7 @@ public void testAlterFilesetProperties() throws IOException { final String filesetName = "fileset" + testId; final String schemaPath = TEST_ROOT_PATH + "/" + schemaName; - createSchema(testId, schemaName, comment, null, schemaPath); + createSchema(schemaName, comment, null, schemaPath); Fileset fileset = createFileset(filesetName, schemaName, comment, Fileset.Type.MANAGED, null, null); @@ -1270,7 +1338,7 @@ public void testUpdateFilesetComment() throws IOException { final String name = "fileset" + testId; final String schemaPath = TEST_ROOT_PATH + "/" + schemaName; - createSchema(testId, schemaName, comment, null, schemaPath); + createSchema(schemaName, comment, null, schemaPath); Fileset fileset = createFileset(name, schemaName, comment, Fileset.Type.MANAGED, null, null); FilesetChange change1 = FilesetChange.updateComment(comment + "_new"); @@ -1294,7 +1362,7 @@ public void testRemoveFilesetComment() throws IOException { final String filesetName = "fileset" + testId; final String schemaPath = TEST_ROOT_PATH + "/" + schemaName; - createSchema(testId, schemaName, comment, null, schemaPath); + createSchema(schemaName, comment, null, schemaPath); Fileset fileset = createFileset(filesetName, schemaName, comment, Fileset.Type.MANAGED, null, null); @@ -1368,7 +1436,7 @@ public void testGetFileLocation() throws IOException { final String storageLocation = TEST_ROOT_PATH + "/" + catalogName + "/" + schemaName + "/" + filesetName; - createSchema(testId, schemaName, comment, null, schemaPath); + createSchema(schemaName, comment, null, schemaPath); Fileset fileset = createFileset( filesetName, schemaName, comment, Fileset.Type.MANAGED, null, storageLocation); @@ -1532,7 +1600,7 @@ public void testCreateSchemaWithDifferentUser() throws Exception { final String schemaName = "schema" + testId; final String comment = "comment" + testId; final String schemaPath = TEST_ROOT_PATH + "/" + schemaName; - return createSchema(testId, schemaName, comment, null, schemaPath, false); + return createSchema(schemaName, comment, null, schemaPath, false); }); Assertions.assertNotNull(schemaCreatedByAlice); @@ -1548,7 +1616,7 @@ public void testCreateSchemaWithDifferentUser() throws Exception { final String comment = "comment" + testId; final String schemaPath = TEST_ROOT_PATH + "/" + schemaName; // Create schema with user "bob" - return createSchema(testId, schemaName, comment, null, schemaPath, false); + return createSchema(schemaName, comment, null, schemaPath, false); }); Assertions.assertNotNull(schemaCreatedByBob); Assertions.assertEquals("bob", schemaCreatedByBob.auditInfo().creator()); @@ -1563,7 +1631,7 @@ public void testCreateSchemaWithDifferentUser() throws Exception { final String comment = "comment" + testId; final String schemaPath = TEST_ROOT_PATH + "/" + schemaName; // Create schema with user "lucy" - return createSchema(testId, schemaName, comment, null, schemaPath, false); + return createSchema(schemaName, comment, null, schemaPath, false); }); Assertions.assertNotNull(schemaCreatedByLucy); Assertions.assertEquals("lucy", schemaCreatedByLucy.auditInfo().creator()); @@ -1577,7 +1645,7 @@ public void testLocationPlaceholdersWithException() throws IOException { final String filesetName = "fileset" + testId; String storageLocation = TEST_ROOT_PATH + "/{{fileset}}/{{user}}/{{id}}"; - createSchema(testId, schemaName, null, null, null); + createSchema(schemaName, null, null, null); Exception exception = Assertions.assertThrows( @@ -1634,7 +1702,7 @@ public void testPlaceholdersInLocation( try (FilesetCatalogOperations ops = new FilesetCatalogOperations(store, secretManager)) { ops.initialize(catalogProps, randomCatalogInfo("m1", "c1"), FILESET_PROPERTIES_METADATA); if (!ops.schemaExists(schemaIdent)) { - createSchema(generateTestId(), schemaName, comment, catalogPath, schemaPath); + createSchema(schemaName, comment, catalogPath, schemaPath); } Fileset fileset = createFileset( @@ -2949,22 +3017,17 @@ private static Stream testRenameArguments() { TEST_ROOT_PATH + "/fileset39")); } - private Schema createSchema( - long testId, String name, String comment, String catalogPath, String schemaPath) + private Schema createSchema(String name, String comment, String catalogPath, String schemaPath) throws IOException { - return createSchema(testId, name, comment, catalogPath, schemaPath, false); + return createSchema(name, comment, catalogPath, schemaPath, false); } private Schema createSchema( - long testId, - String name, - String comment, - String catalogPath, - String schemaPath, - boolean disableFsOps) + String name, String comment, String catalogPath, String schemaPath, boolean disableFsOps) throws IOException { + long schemaId = idGenerator.nextId(); // stub schema - doReturn(new SchemaIds(1L, 1L, testId)) + doReturn(new SchemaIds(1L, 1L, schemaId)) .when(spySchemaMetaService) .getSchemaIdByMetalakeNameAndCatalogNameAndSchemaName( Mockito.anyString(), Mockito.anyString(), Mockito.eq(name)); @@ -2980,7 +3043,7 @@ private Schema createSchema( NameIdentifier schemaIdent = NameIdentifierUtil.ofSchema("m1", "c1", name); Map schemaProps = Maps.newHashMap(); - StringIdentifier stringId = StringIdentifier.fromId(testId); + StringIdentifier stringId = StringIdentifier.fromId(schemaId); schemaProps = Maps.newHashMap(StringIdentifier.newPropertiesWithId(stringId, schemaProps)); if (schemaPath != null) { diff --git a/catalogs/catalog-kafka/src/test/java/org/apache/gravitino/catalog/kafka/TestKafkaCatalogOperations.java b/catalogs/catalog-kafka/src/test/java/org/apache/gravitino/catalog/kafka/TestKafkaCatalogOperations.java index 16b49512361..fd21391cb93 100644 --- a/catalogs/catalog-kafka/src/test/java/org/apache/gravitino/catalog/kafka/TestKafkaCatalogOperations.java +++ b/catalogs/catalog-kafka/src/test/java/org/apache/gravitino/catalog/kafka/TestKafkaCatalogOperations.java @@ -44,7 +44,6 @@ import static org.apache.gravitino.catalog.kafka.KafkaCatalogPropertiesMetadata.BOOTSTRAP_SERVERS; import static org.apache.gravitino.catalog.kafka.KafkaTopicPropertiesMetadata.PARTITION_COUNT; import static org.apache.gravitino.catalog.kafka.KafkaTopicPropertiesMetadata.REPLICATION_FACTOR; -import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.when; import com.google.common.collect.ImmutableMap; @@ -72,18 +71,16 @@ import org.apache.gravitino.messaging.Topic; import org.apache.gravitino.messaging.TopicChange; import org.apache.gravitino.meta.AuditInfo; +import org.apache.gravitino.meta.BaseMetalake; import org.apache.gravitino.meta.CatalogEntity; +import org.apache.gravitino.meta.SchemaVersion; import org.apache.gravitino.storage.IdGenerator; import org.apache.gravitino.storage.RandomIdGenerator; -import org.apache.gravitino.storage.relational.helper.CatalogIds; -import org.apache.gravitino.storage.relational.service.CatalogMetaService; -import org.apache.gravitino.storage.relational.service.MetalakeMetaService; import org.apache.kafka.common.config.TopicConfig; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; -import org.mockito.MockedStatic; import org.mockito.Mockito; public class TestKafkaCatalogOperations extends KafkaClusterEmbedded { @@ -139,7 +136,7 @@ public PropertiesMetadata modelVersionPropertiesMetadata() private static KafkaCatalogOperations kafkaCatalogOperations; @BeforeAll - public static void setUp() throws IllegalAccessException { + public static void setUp() throws IOException, IllegalAccessException { Config config = Mockito.mock(Config.class); Mockito.when(config.get(STORE_TRANSACTION_MAX_SKEW_TIME)).thenReturn(1000L); Mockito.when(config.get(STORE_DELETE_AFTER_TIME)).thenReturn(20 * 60 * 1000L); @@ -174,35 +171,23 @@ public static void setUp() throws IllegalAccessException { Mockito.when(config.get(Configs.CACHE_IMPLEMENTATION)).thenReturn("caffeine"); Mockito.when(config.get(Configs.CACHE_LOCK_SEGMENTS)).thenReturn(16); - // Mock - MetalakeMetaService metalakeMetaService = MetalakeMetaService.getInstance(); - MetalakeMetaService spyMetaservice = Mockito.spy(metalakeMetaService); - doReturn(1L).when(spyMetaservice).getMetalakeIdByName(Mockito.anyString()); - - CatalogMetaService catalogMetaService = CatalogMetaService.getInstance(); - CatalogMetaService spyCatalogMetaService = Mockito.spy(catalogMetaService); - doReturn(1L) - .when(spyCatalogMetaService) - .getCatalogIdByMetalakeIdAndName(Mockito.anyLong(), Mockito.anyString()); - doReturn(new CatalogIds(1L, 1L)) - .when(spyCatalogMetaService) - .getCatalogIdByMetalakeAndCatalogName(Mockito.anyString(), Mockito.anyString()); - - MockedStatic metalakeMetaServiceMockedStatic = - Mockito.mockStatic(MetalakeMetaService.class); - MockedStatic catalogMetaServiceMockedStatic = - Mockito.mockStatic(CatalogMetaService.class); - - metalakeMetaServiceMockedStatic - .when(MetalakeMetaService::getInstance) - .thenReturn(spyMetaservice); - catalogMetaServiceMockedStatic - .when(CatalogMetaService::getInstance) - .thenReturn(spyCatalogMetaService); - store = EntityStoreFactory.createEntityStore(config); store.initialize(config); idGenerator = new RandomIdGenerator(); + + BaseMetalake metalake = + BaseMetalake.builder() + .withId(1L) + .withName(METALAKE_NAME) + .withVersion(SchemaVersion.V_0_1) + .withAuditInfo( + AuditInfo.builder() + .withCreator("testKafkaUser") + .withCreateTime(Instant.now()) + .build()) + .build(); + store.put(metalake, false); + kafkaCatalogEntity = CatalogEntity.builder() .withId(1L) @@ -217,6 +202,7 @@ public static void setUp() throws IllegalAccessException { .withCreateTime(Instant.now()) .build()) .build(); + store.put(kafkaCatalogEntity, false); FieldUtils.writeField(GravitinoEnv.getInstance(), "config", config, true); @@ -234,11 +220,11 @@ public static void tearDown() throws IOException { } @Test - public void testKafkaCatalogConfiguration() { + public void testKafkaCatalogConfiguration() throws IOException { String catalogName = "test_kafka_catalog_configuration"; CatalogEntity catalogEntity = CatalogEntity.builder() - .withId(2L) + .withId(idGenerator.nextId()) .withName(catalogName) .withNamespace(Namespace.of(METALAKE_NAME)) .withType(MESSAGING) @@ -250,6 +236,7 @@ public void testKafkaCatalogConfiguration() { .build()) .withProperties(MOCK_CATALOG_PROPERTIES) .build(); + store.put(catalogEntity, false); KafkaCatalogOperations ops = new KafkaCatalogOperations(store, idGenerator); Assertions.assertNull(ops.adminClientConfig); @@ -270,11 +257,11 @@ public void testKafkaCatalogConfiguration() { } @Test - public void testInitialization() { + public void testInitialization() throws IOException { String catalogName = "test_kafka_catalog_initialization"; CatalogEntity catalogEntity = CatalogEntity.builder() - .withId(2L) + .withId(idGenerator.nextId()) .withName(catalogName) .withNamespace(Namespace.of(METALAKE_NAME)) .withType(MESSAGING) @@ -286,6 +273,7 @@ public void testInitialization() { .build()) .withProperties(MOCK_CATALOG_PROPERTIES) .build(); + store.put(catalogEntity, false); KafkaCatalogOperations ops = new KafkaCatalogOperations(store, idGenerator); ops.initialize( MOCK_CATALOG_PROPERTIES, catalogEntity.toCatalogInfo(), KAFKA_PROPERTIES_METADATA); diff --git a/core/src/main/java/org/apache/gravitino/catalog/ManagedSchemaOperations.java b/core/src/main/java/org/apache/gravitino/catalog/ManagedSchemaOperations.java index 164d4b53460..1c2dbade53d 100644 --- a/core/src/main/java/org/apache/gravitino/catalog/ManagedSchemaOperations.java +++ b/core/src/main/java/org/apache/gravitino/catalog/ManagedSchemaOperations.java @@ -25,6 +25,7 @@ import java.util.List; import java.util.Map; import org.apache.gravitino.Entity; +import org.apache.gravitino.EntityAlreadyExistsException; import org.apache.gravitino.EntityStore; import org.apache.gravitino.NameIdentifier; import org.apache.gravitino.Namespace; @@ -116,9 +117,13 @@ public Schema createSchema(NameIdentifier ident, String comment, Map schemaIds); + /** + * Soft-deletes a schema, but only while it still carries the given version. + * + * @param schemaId the ID of the schema to delete + * @param currentVersion the version the caller read before deciding to delete + * @return 1 when the schema was deleted, 0 when it changed or is already gone + */ + @UpdateProvider( + type = SchemaMetaSQLProviderFactory.class, + method = "softDeleteSchemaMetaBySchemaIdAndVersion") + Integer softDeleteSchemaMetaBySchemaIdAndVersion( + @Param("schemaId") Long schemaId, @Param("currentVersion") Long currentVersion); + /** * Soft-deletes schemas whose identifiers and OCC versions still match. * 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 557bee15f81..62c532db549 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 @@ -49,7 +49,15 @@ public static SchemaMetaBaseSQLProvider getProvider() { static class SchemaMetaMySQLProvider extends SchemaMetaBaseSQLProvider {} - static class SchemaMetaH2Provider extends SchemaMetaBaseSQLProvider {} + static class SchemaMetaH2Provider extends SchemaMetaBaseSQLProvider { + @Override + public String selectSchemaMetaByIdForShare(Long schemaId) { + // H2 has no shared row-lock syntax, so H2 backends fall back to an exclusive lock. Writes of + // tables, views, filesets and the like under one schema therefore serialize on H2, and a slow + // write can make a concurrent one hit H2's lock timeout instead of a clean conflict. + return selectSchemaMetaByIdForUpdate(schemaId); + } + } public static String listSchemaPOsByFullQualifiedName( @Param("metalakeName") String metalakeName, @Param("catalogName") String catalogName) { @@ -98,6 +106,16 @@ public static String selectSchemaMetaById(@Param("schemaId") Long schemaId) { return getProvider().selectSchemaMetaById(schemaId); } + /** Returns SQL that selects and locks an active schema by ID. */ + public static String selectSchemaMetaByIdForUpdate(@Param("schemaId") Long schemaId) { + return getProvider().selectSchemaMetaByIdForUpdate(schemaId); + } + + /** Returns SQL that selects and share-locks an active schema by ID. */ + public static String selectSchemaMetaByIdForShare(@Param("schemaId") Long schemaId) { + return getProvider().selectSchemaMetaByIdForShare(schemaId); + } + public static String insertSchemaMeta(@Param("schemaMeta") SchemaPO schemaPO) { return getProvider().insertSchemaMeta(schemaPO); } @@ -125,6 +143,11 @@ public static String softDeleteSchemaMetasBySchemaIds(@Param("schemaIds") List schemaPOs) { diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/CatalogMetaBaseSQLProvider.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/CatalogMetaBaseSQLProvider.java index e6f8f03c183..d877aea8397 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/CatalogMetaBaseSQLProvider.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/CatalogMetaBaseSQLProvider.java @@ -147,6 +147,11 @@ public String selectCatalogMetaByIdForUpdate(@Param("catalogId") Long catalogId) return selectCatalogMetaById(catalogId) + " FOR UPDATE"; } + /** Returns SQL that selects and share-locks an active catalog by ID. */ + public String selectCatalogMetaByIdForShare(@Param("catalogId") Long catalogId) { + return selectCatalogMetaById(catalogId) + " LOCK IN SHARE MODE"; + } + public String insertCatalogMeta(@Param("catalogMeta") CatalogPO catalogPO) { return "INSERT INTO " + TABLE_NAME 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 9ee36cc52da..76f989b3b6b 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 @@ -182,6 +182,16 @@ public String selectSchemaMetaById(@Param("schemaId") Long schemaId) { + " WHERE schema_id = #{schemaId} AND deleted_at = 0"; } + /** Returns SQL that selects and locks an active schema by ID. */ + public String selectSchemaMetaByIdForUpdate(@Param("schemaId") Long schemaId) { + return selectSchemaMetaById(schemaId) + " FOR UPDATE"; + } + + /** Returns SQL that selects and share-locks an active schema by ID. */ + public String selectSchemaMetaByIdForShare(@Param("schemaId") Long schemaId) { + return selectSchemaMetaById(schemaId) + " LOCK IN SHARE MODE"; + } + public String insertSchemaMeta(@Param("schemaMeta") SchemaPO schemaPO) { return "INSERT INTO " + TABLE_NAME @@ -227,8 +237,12 @@ public String insertSchemaMetaOnDuplicateKeyUpdate(@Param("schemaMeta") SchemaPO + " schema_comment = #{schemaMeta.schemaComment}," + " properties = #{schemaMeta.properties}," + " audit_info = #{schemaMeta.auditInfo}," - + " current_version = #{schemaMeta.currentVersion}," - + " last_version = #{schemaMeta.lastVersion}," + // Move the version forward instead of writing the initial version again. Resetting it + // would let a slow alter or drop that still holds an older version pass its own version + // check later on. last_version is assigned first, so both columns are computed from the + // version the row had before this statement. + + " last_version = current_version + 1," + + " current_version = current_version + 1," + " deleted_at = #{schemaMeta.deletedAt}"; } @@ -265,12 +279,24 @@ public String batchInsertSchemaMetaOnDuplicateKeyUpdate( + " schema_comment = VALUES(schema_comment)," + " properties = VALUES(properties)," + " audit_info = VALUES(audit_info)," - + " current_version = VALUES(current_version)," - + " last_version = VALUES(last_version)," + // Move the version forward instead of writing the initial version again. Resetting it + // would let a slow alter or drop that still holds an older version pass its own version + // check later on. last_version is assigned first, so both columns are computed from the + // version the row had before this statement. + + " last_version = current_version + 1," + + " current_version = current_version + 1," + " deleted_at = VALUES(deleted_at)" + ""; } + /** + * Builds SQL that updates a schema only if nobody changed it in the meantime. + * + *

The WHERE clause used to repeat every column. Comparing the version alone is enough now, + * because every update moves the version forward, and it also avoids a MySQL trap: MySQL reports + * zero affected rows when an UPDATE writes the values a row already has, which the old SQL could + * not tell apart from a real conflict. + */ public String updateSchemaMeta( @Param("newSchemaMeta") SchemaPO newSchemaPO, @Param("oldSchemaMeta") SchemaPO oldSchemaPO) { return "UPDATE " @@ -285,15 +311,7 @@ public String updateSchemaMeta( + " last_version = #{newSchemaMeta.lastVersion}," + " deleted_at = #{newSchemaMeta.deletedAt}" + " WHERE schema_id = #{oldSchemaMeta.schemaId}" - + " AND schema_name = #{oldSchemaMeta.schemaName}" - + " AND metalake_id = #{oldSchemaMeta.metalakeId}" - + " AND catalog_id = #{oldSchemaMeta.catalogId}" - + " AND (schema_comment = #{oldSchemaMeta.schemaComment}" - + " OR (schema_comment IS NULL and #{oldSchemaMeta.schemaComment} IS NULL))" - + " AND properties = #{oldSchemaMeta.properties}" - + " AND audit_info = #{oldSchemaMeta.auditInfo}" + " AND current_version = #{oldSchemaMeta.currentVersion}" - + " AND last_version = #{oldSchemaMeta.lastVersion}" + " AND deleted_at = 0"; } @@ -311,6 +329,16 @@ public String softDeleteSchemaMetasBySchemaIds(@Param("schemaIds") List sc + ""; } + public String softDeleteSchemaMetaBySchemaIdAndVersion( + @Param("schemaId") Long schemaId, @Param("currentVersion") Long currentVersion) { + return "UPDATE " + + TABLE_NAME + + " SET deleted_at = (UNIX_TIMESTAMP() * 1000.0)" + + " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000" + + " WHERE schema_id = #{schemaId}" + + " AND current_version = #{currentVersion} AND deleted_at = 0"; + } + /** Returns SQL that soft-deletes schemas using identifier-and-version pairs. */ public String softDeleteSchemaMetasWithVersion(@Param("schemaMetas") List schemaPOs) { return ""; } @@ -98,16 +121,7 @@ public String updateSchemaMeta( + " last_version = #{newSchemaMeta.lastVersion}," + " deleted_at = #{newSchemaMeta.deletedAt}" + " WHERE schema_id = #{oldSchemaMeta.schemaId}" - + " AND schema_name = #{oldSchemaMeta.schemaName}" - + " AND metalake_id = #{oldSchemaMeta.metalakeId}" - + " AND catalog_id = #{oldSchemaMeta.catalogId}" - + " AND (schema_comment = #{oldSchemaMeta.schemaComment}" - + " OR (CAST(schema_comment AS VARCHAR) IS NULL" - + " AND CAST(#{oldSchemaMeta.schemaComment} AS VARCHAR) IS NULL))" - + " AND properties = #{oldSchemaMeta.properties}" - + " AND audit_info = #{oldSchemaMeta.auditInfo}" + " AND current_version = #{oldSchemaMeta.currentVersion}" - + " AND last_version = #{oldSchemaMeta.lastVersion}" + " AND deleted_at = 0"; } @@ -125,6 +139,15 @@ public String softDeleteSchemaMetasBySchemaIds(List schemaIds) { + ""; } + @Override + public String softDeleteSchemaMetaBySchemaIdAndVersion(Long schemaId, Long currentVersion) { + return "UPDATE " + + TABLE_NAME + + " SET deleted_at = CAST(EXTRACT(EPOCH FROM CURRENT_TIMESTAMP) * 1000 AS BIGINT)" + + " WHERE schema_id = #{schemaId}" + + " AND current_version = #{currentVersion} AND deleted_at = 0"; + } + /** {@inheritDoc} */ @Override public String softDeleteSchemaMetasWithVersion(List schemaPOs) { diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/service/CatalogMetaService.java b/core/src/main/java/org/apache/gravitino/storage/relational/service/CatalogMetaService.java index 5cd9ca66e3a..8f32ac0563c 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/service/CatalogMetaService.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/service/CatalogMetaService.java @@ -505,9 +505,7 @@ private RuntimeException catalogWriteFailure( * must already hold the catalog row, so no schema can appear or disappear in between. */ private void deleteSchemasWithVersions(NameIdentifier catalogIdentifier, Long catalogId) { - List schemaPOs = - SessionUtils.getWithoutCommit( - SchemaMetaMapper.class, mapper -> mapper.listSchemaPOsByCatalogId(catalogId)); + List schemaPOs = listSchemaPOsForCascade(catalogId); if (schemaPOs.isEmpty()) { return; } @@ -521,4 +519,14 @@ private void deleteSchemasWithVersions(NameIdentifier catalogIdentifier, Long ca Entity.EntityType.SCHEMA, Entity.EntityType.CATALOG, catalogIdentifier); } } + + /** + * Reads the schemas that the cascade is about to delete. The caller already holds the catalog + * row, so this snapshot cannot grow or shrink behind it. Kept separate so a test can pause the + * cascade exactly here, between taking the lock and reading the children. + */ + List listSchemaPOsForCascade(Long catalogId) { + return SessionUtils.getWithoutCommit( + SchemaMetaMapper.class, mapper -> mapper.listSchemaPOsByCatalogId(catalogId)); + } } 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 4dbcadbe383..b29d1981931 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 @@ -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( + filesetEntity.nameIdentifier(), + po.getSchemaId(), + po.getCatalogId(), + po.getMetalakeId()), () -> SessionUtils.doWithoutCommit( FilesetMetaMapper.class, 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 2c582dc8c0d..04976bed87a 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 @@ -115,6 +115,15 @@ public void insertFunction(FunctionEntity functionEntity, boolean overwrite) thr 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)), @@ -263,14 +272,32 @@ public FunctionEntity updateFunction( 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())), - () -> - SessionUtils.doWithoutCommit( - FunctionMetaMapper.class, - mapper -> ops.updatePO(mapper, newFunctionPO, oldFunctionPO))); + () -> { + 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); + } + }); 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 838a17fdefc..00ef4aecea3 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 @@ -96,17 +96,28 @@ public void insertModel(ModelEntity modelEntity, boolean overwrite) throws IOExc try { ModelPO.Builder builder = ModelPO.builder(); fillModelPOBuilderParentEntityId(builder, modelEntity.namespace()); + ModelPO po = POConverters.initializeModelPO(modelEntity, builder); - SessionUtils.doWithCommit( - ModelMetaMapper.class, - mapper -> { - ModelPO po = POConverters.initializeModelPO(modelEntity, builder); - if (overwrite) { - mapper.insertModelMetaOnDuplicateKeyUpdate(po); - } else { - mapper.insertModelMeta(po); - } - }); + 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); + } + })); } catch (RuntimeException re) { ExceptionUtils.checkSQLException( re, Entity.EntityType.MODEL, modelEntity.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 38dd40576a8..b9b90f5d942 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 @@ -46,6 +46,7 @@ import org.apache.gravitino.storage.relational.mapper.ModelMetaMapper; import org.apache.gravitino.storage.relational.mapper.ModelVersionAliasRelMapper; import org.apache.gravitino.storage.relational.mapper.ModelVersionMetaMapper; +import org.apache.gravitino.storage.relational.po.ModelPO; import org.apache.gravitino.storage.relational.po.ModelVersionAliasRelPO; import org.apache.gravitino.storage.relational.po.ModelVersionPO; import org.apache.gravitino.storage.relational.utils.ExceptionUtils; @@ -162,7 +163,8 @@ public void insertModelVersion(ModelVersionEntity modelVersionEntity) throws IOE NameIdentifier modelIdent = modelVersionEntity.modelIdentifier(); NameIdentifierUtil.checkModel(modelIdent); - Long modelId = EntityIdService.getEntityId(modelIdent, Entity.EntityType.MODEL); + ModelPO modelPO = ModelMetaService.getInstance().getModelPOByIdentifier(modelIdent); + Long modelId = modelPO.getModelId(); List modelVersionPOs = POConverters.initializeModelVersionPO(modelVersionEntity, modelId); @@ -171,6 +173,10 @@ public void insertModelVersion(ModelVersionEntity modelVersionEntity) throws IOE 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), () -> SessionUtils.doWithoutCommit( ModelVersionMetaMapper.class, @@ -183,10 +189,17 @@ public void insertModelVersion(ModelVersionEntity modelVersionEntity) throws IOE ModelVersionAliasRelMapper.class, mapper -> mapper.insertModelVersionAliasRels(aliasRelPOs)); }, - () -> - // If the model version is inserted successfully, update the model latest version. - SessionUtils.doWithoutCommit( - ModelMetaMapper.class, mapper -> mapper.updateModelLatestVersion(modelId))); + () -> { + // 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. + int updated = + SessionUtils.getWithoutCommit( + ModelMetaMapper.class, mapper -> mapper.updateModelLatestVersion(modelId)); + if (updated == 0) { + throw noSuchModelException(modelIdent); + } + }); } catch (RuntimeException re) { ExceptionUtils.checkSQLException( re, Entity.EntityType.MODEL_VERSION, modelVersionEntity.modelIdentifier().toString()); @@ -291,17 +304,17 @@ public ModelVersionEntity updateModelVersion( NameIdentifier modelIdent = NameIdentifier.of(ident.namespace().levels()); boolean isVersionNumber = NumberUtils.isCreatable(ident.name()); - ModelEntity modelEntity = ModelMetaService.getInstance().getModelByIdentifier(modelIdent); + ModelPO modelPO = ModelMetaService.getInstance().getModelPOByIdentifier(modelIdent); + Long modelId = modelPO.getModelId(); List oldModelVersionPOs = SessionUtils.getWithoutCommit( ModelVersionMetaMapper.class, mapper -> { if (isVersionNumber) { - return mapper.selectModelVersionMeta( - modelEntity.id(), Integer.valueOf(ident.name())); + return mapper.selectModelVersionMeta(modelId, Integer.valueOf(ident.name())); } else { - return mapper.selectModelVersionMetaByAlias(modelEntity.id(), ident.name()); + return mapper.selectModelVersionMetaByAlias(modelId, ident.name()); } }); @@ -318,10 +331,9 @@ public ModelVersionEntity updateModelVersion( mapper -> { if (isVersionNumber) { return mapper.selectModelVersionAliasRelsByModelIdAndVersion( - modelEntity.id(), Integer.valueOf(ident.name())); + modelId, Integer.valueOf(ident.name())); } else { - return mapper.selectModelVersionAliasRelsByModelIdAndAlias( - modelEntity.id(), ident.name()); + return mapper.selectModelVersionAliasRelsByModelIdAndAlias(modelId, ident.name()); } }); @@ -339,8 +351,7 @@ public ModelVersionEntity updateModelVersion( boolean isAliasChanged = isModelVersionAliasUpdated(oldModelVersionEntity, newModelVersionEntity); List newAliasRelPOs = - POConverters.updateModelVersionAliasRelPO( - oldAliasRelPOs, newModelVersionEntity, modelEntity.id()); + POConverters.updateModelVersionAliasRelPO(oldAliasRelPOs, newModelVersionEntity, modelId); boolean isModelVersionUriUpdated = isModelVersionUriUpdated(oldModelVersionEntity, newModelVersionEntity); @@ -348,6 +359,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), () -> { if (isModelVersionUriUpdated) { // delete old model version POs first @@ -357,16 +371,16 @@ public ModelVersionEntity updateModelVersion( mapper -> { if (isVersionNumber) { return mapper.softDeleteModelVersionMetaByModelIdAndVersion( - modelEntity.id(), Integer.valueOf(ident.name())); + modelId, Integer.valueOf(ident.name())); } else { return mapper.softDeleteModelVersionMetaByModelIdAndAlias( - modelEntity.id(), ident.name()); + modelId, ident.name()); } })); // insert model version POs with updated URIs List modelVersionPOs = - POConverters.initializeModelVersionPO(newModelVersionEntity, modelEntity.id()); + POConverters.initializeModelVersionPO(newModelVersionEntity, modelId); SessionUtils.doWithoutCommit( ModelVersionMetaMapper.class, mapper -> mapper.insertModelVersionMetasWithVersionNumber(modelVersionPOs)); @@ -392,7 +406,7 @@ public ModelVersionEntity updateModelVersion( .forEach( alias -> mapper.softDeleteModelVersionAliasRelsByModelIdAndAlias( - modelEntity.id(), alias))); + modelId, alias))); SessionUtils.doWithoutCommit( ModelVersionAliasRelMapper.class, @@ -431,4 +445,21 @@ private boolean isModelVersionUriUpdated( Map newUris = newModelVersionEntity.uris(); return !oldUris.equals(newUris); } + + private void lockSchemaForModelVersionWrite( + NameIdentifier modelIdentifier, ModelPO observedModelPO) { + SchemaMetaService.getInstance() + .lockSchemaForEntityWrite( + modelIdentifier, + observedModelPO.getSchemaId(), + observedModelPO.getCatalogId(), + observedModelPO.getMetalakeId()); + } + + private NoSuchEntityException noSuchModelException(NameIdentifier modelIdentifier) { + return new NoSuchEntityException( + NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE, + Entity.EntityType.MODEL.name().toLowerCase(Locale.ROOT), + modelIdentifier.toString()); + } } 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 898687f80f8..40eacb0b715 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 @@ -26,9 +26,9 @@ import java.util.Arrays; import java.util.Collections; import java.util.List; +import java.util.Map; import java.util.Objects; -import java.util.Set; -import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import java.util.function.Function; import java.util.regex.Pattern; import java.util.stream.Collectors; @@ -40,15 +40,11 @@ import org.apache.gravitino.Namespace; import org.apache.gravitino.exceptions.NoSuchEntityException; import org.apache.gravitino.exceptions.NonEmptyEntityException; -import org.apache.gravitino.meta.FilesetEntity; -import org.apache.gravitino.meta.ModelEntity; -import org.apache.gravitino.meta.NamespacedEntityId; import org.apache.gravitino.meta.SchemaEntity; -import org.apache.gravitino.meta.TableEntity; -import org.apache.gravitino.meta.TopicEntity; import org.apache.gravitino.metrics.Monitored; import org.apache.gravitino.storage.IdGenerator; import org.apache.gravitino.storage.relational.helper.SchemaIds; +import org.apache.gravitino.storage.relational.mapper.CatalogMetaMapper; import org.apache.gravitino.storage.relational.mapper.FilesetMetaMapper; import org.apache.gravitino.storage.relational.mapper.FilesetVersionMapper; import org.apache.gravitino.storage.relational.mapper.FunctionMetaMapper; @@ -66,6 +62,7 @@ import org.apache.gravitino.storage.relational.mapper.TagMetadataObjectRelMapper; import org.apache.gravitino.storage.relational.mapper.TopicMetaMapper; import org.apache.gravitino.storage.relational.mapper.ViewMetaMapper; +import org.apache.gravitino.storage.relational.po.CatalogPO; import org.apache.gravitino.storage.relational.po.SchemaPO; import org.apache.gravitino.storage.relational.utils.ExceptionUtils; import org.apache.gravitino.storage.relational.utils.POConverters; @@ -143,6 +140,10 @@ public void insertSchema(SchemaEntity schemaEntity, boolean overwrite) throws IO // rewriter to translate each PO's name to storage form before SQL execution. String logicalSep = HierarchicalSchemaUtil.schemaSeparator(); String schemaName = schemaEntity.name(); + String metalakeName = schemaEntity.namespace().level(0); + String catalogName = schemaEntity.namespace().level(1); + CatalogPO catalogPO = + CatalogMetaService.getInstance().getCatalogPOByName(metalakeName, catalogName); List rowsToInsert = new ArrayList<>(); if (schemaName == null || !schemaName.contains(logicalSep)) { rowsToInsert.add(schemaEntity); @@ -165,39 +166,55 @@ public void insertSchema(SchemaEntity schemaEntity, boolean overwrite) throws IO rowsToInsert.add(schemaEntity); } - SessionUtils.doWithCommit( - SchemaMetaMapper.class, - mapper -> { - int n = rowsToInsert.size(); - List missingAncestorPOs = new ArrayList<>(); - if (n > 1) { - SchemaEntity firstAncestor = rowsToInsert.get(0); - Namespace ancestorNs = firstAncestor.namespace(); - List ancestorNames = - rowsToInsert.subList(0, n - 1).stream() - .map(SchemaEntity::name) - .collect(Collectors.toList()); - Set existingLogicalNames = - ops.listPOs(mapper, ancestorNs, ancestorNames).stream() - .map(SchemaPO::getSchemaName) - .collect(Collectors.toSet()); - for (SchemaEntity row : rowsToInsert.subList(0, n - 1)) { - if (existingLogicalNames.contains(row.name())) { - continue; - } - SchemaPO.Builder builder = SchemaPO.builder(); - fillSchemaPOBuilderParentEntityId(builder, row.namespace()); - missingAncestorPOs.add(POConverters.initializeSchemaPOWithVersion(row, builder)); - } - } - SchemaEntity leafRow = rowsToInsert.get(n - 1); - SchemaPO.Builder leafBuilder = SchemaPO.builder(); - fillSchemaPOBuilderParentEntityId(leafBuilder, leafRow.namespace()); - SchemaPO leafPO = POConverters.initializeSchemaPOWithVersion(leafRow, leafBuilder); - List schemaPosToInsert = new ArrayList<>(missingAncestorPOs); - schemaPosToInsert.add(leafPO); - ops.batchInsertPOs(mapper, schemaPosToInsert, overwrite); - }); + // Everything below runs in one transaction, and it starts by locking the parent catalog row. + // That lock is what stops a catalog drop from running at the same time as this insert. A + // plain name is enough with a shared lock; a nested name needs an exclusive one, see + // lockCatalogForSchemaCreate. + SessionUtils.doMultipleWithCommit( + () -> lockCatalogForSchemaCreate(catalogPO, rowsToInsert.size() > 1), + () -> + SessionUtils.doWithoutCommit( + SchemaMetaMapper.class, + mapper -> { + int n = rowsToInsert.size(); + List missingAncestorPOs = new ArrayList<>(); + if (n > 1) { + // Only insert the ancestors that are not there yet. Reading them inside the + // transaction is safe because the exclusive catalog lock is already held, so + // no other request can add the same ancestor between this read and the + // insert below. + SchemaEntity firstAncestor = rowsToInsert.get(0); + Namespace ancestorNs = firstAncestor.namespace(); + List ancestorNames = + rowsToInsert.subList(0, n - 1).stream() + .map(SchemaEntity::name) + .collect(Collectors.toList()); + Map existingAncestors = + ops.listPOs(mapper, ancestorNs, ancestorNames).stream() + .collect( + Collectors.toMap(SchemaPO::getSchemaName, Function.identity())); + for (SchemaEntity row : rowsToInsert.subList(0, n - 1)) { + SchemaPO existingAncestor = existingAncestors.get(row.name()); + if (existingAncestor != null) { + continue; + } + SchemaPO.Builder builder = newSchemaPOBuilder(catalogPO); + missingAncestorPOs.add( + POConverters.initializeSchemaPOWithVersion(row, builder)); + } + } + if (!missingAncestorPOs.isEmpty()) { + ops.batchInsertPOs(mapper, missingAncestorPOs, false); + } + // The schema the caller actually asked for. Ancestors above are filled in + // silently, but this row must obey the caller's choice: with overwrite off, a + // name that is already taken fails instead of replacing the existing schema. + SchemaEntity leafRow = rowsToInsert.get(n - 1); + SchemaPO leafPO = + POConverters.initializeSchemaPOWithVersion( + leafRow, newSchemaPOBuilder(catalogPO)); + ops.batchInsertPOs(mapper, Collections.singletonList(leafPO), overwrite); + })); } catch (RuntimeException re) { ExceptionUtils.checkSQLException( re, Entity.EntityType.SCHEMA, schemaEntity.nameIdentifier().toString()); @@ -219,29 +236,33 @@ public SchemaEntity updateSchema( newEntity.id(), oldSchemaEntity.id()); - AtomicInteger updateResult = new AtomicInteger(0); try { SessionUtils.doMultipleWithCommit( - () -> - updateResult.set( - SessionUtils.getWithoutCommit( - SchemaMetaMapper.class, - mapper -> - ops.updatePO( - mapper, - POConverters.updateSchemaPOWithVersion(oldSchemaPO, newEntity), - oldSchemaPO)))); + () -> { + // The UPDATE only matches the row while it still carries the version read above, and it + // writes the next version. Two servers that started from the same schema therefore + // cannot both apply their change: the slower one updates no row. + int updated = + SessionUtils.getWithoutCommit( + SchemaMetaMapper.class, + mapper -> + ops.updatePO( + mapper, + POConverters.updateSchemaPOWithVersion(oldSchemaPO, newEntity), + oldSchemaPO)); + if (updated == 0) { + // Zero rows has two possible causes: someone else changed the schema, or the schema + // is gone. schemaWriteFailure tells them apart and picks the right error. + throw schemaWriteFailure(identifier, oldSchemaPO); + } + }); } catch (RuntimeException re) { ExceptionUtils.checkSQLException( re, Entity.EntityType.SCHEMA, newEntity.nameIdentifier().toString()); throw re; } - if (updateResult.get() > 0) { - return newEntity; - } else { - throw new IOException("Failed to update the entity: " + identifier); - } + return newEntity; } @Monitored( @@ -250,140 +271,102 @@ public SchemaEntity updateSchema( public boolean deleteSchema(NameIdentifier identifier, boolean cascade) { NameIdentifierUtil.checkSchema(identifier); - String schemaName = identifier.name(); SchemaPO schemaPO = getSchemaPOByIdentifier(identifier); Long schemaId = schemaPO.getSchemaId(); if (cascade) { - // For HierarchicalSchema, deleting `A:B` must also cascade into all descendant schemas - // such as `A:B:C`, `A:B:C:D`, etc. Collect the descendant schema ids up-front and run a - // single batch UPDATE per child table so the total SQL cost stays bounded regardless of - // how many descendants exist. - List schemaIds = listSchemaIdsForCascade(schemaPO); - if (schemaIds.isEmpty()) { - return false; - } + AtomicReference> schemaIds = new AtomicReference<>(); SessionUtils.doMultipleWithCommit( - () -> - SessionUtils.doWithoutCommit( - SchemaMetaMapper.class, - mapper -> mapper.softDeleteSchemaMetasBySchemaIds(schemaIds)), + () -> { + // Take the parent catalog lock first, then delete this schema, and only then look at + // its descendants. Schema creation takes the same catalog lock, so once we hold it no + // schema can appear or disappear under us. Every overlapping drop grabs the locks in + // this same order, which is what keeps two cascades from deadlocking each other. + lockCatalogForSchemaDelete(identifier, schemaPO); + deleteSchemaWithVersion(identifier, schemaPO); + List descendants = listDescendantSchemaPOs(schemaPO); + deleteDescendantSchemasWithVersions(identifier, descendants); + List ids = new ArrayList<>(descendants.size() + 1); + ids.add(schemaId); + descendants.stream().map(SchemaPO::getSchemaId).forEach(ids::add); + schemaIds.set(ids); + }, () -> SessionUtils.doWithoutCommit( TableMetaMapper.class, - mapper -> mapper.softDeleteTableMetasBySchemaIds(schemaIds)), + mapper -> mapper.softDeleteTableMetasBySchemaIds(schemaIds.get())), () -> SessionUtils.doWithoutCommit( TableColumnMapper.class, - mapper -> mapper.softDeleteColumnsBySchemaIds(schemaIds)), + mapper -> mapper.softDeleteColumnsBySchemaIds(schemaIds.get())), () -> SessionUtils.doWithoutCommit( FilesetMetaMapper.class, - mapper -> mapper.softDeleteFilesetMetasBySchemaIds(schemaIds)), + mapper -> mapper.softDeleteFilesetMetasBySchemaIds(schemaIds.get())), () -> SessionUtils.doWithoutCommit( FilesetVersionMapper.class, - mapper -> mapper.softDeleteFilesetVersionsBySchemaIds(schemaIds)), + mapper -> mapper.softDeleteFilesetVersionsBySchemaIds(schemaIds.get())), () -> SessionUtils.doWithoutCommit( TopicMetaMapper.class, - mapper -> mapper.softDeleteTopicMetasBySchemaIds(schemaIds)), + mapper -> mapper.softDeleteTopicMetasBySchemaIds(schemaIds.get())), () -> SessionUtils.doWithoutCommit( FunctionMetaMapper.class, - mapper -> mapper.softDeleteFunctionMetasBySchemaIds(schemaIds)), + mapper -> mapper.softDeleteFunctionMetasBySchemaIds(schemaIds.get())), () -> SessionUtils.doWithoutCommit( FunctionVersionMetaMapper.class, - mapper -> mapper.softDeleteFunctionVersionMetasBySchemaIds(schemaIds)), + mapper -> mapper.softDeleteFunctionVersionMetasBySchemaIds(schemaIds.get())), () -> SessionUtils.doWithoutCommit( - OwnerMetaMapper.class, mapper -> mapper.softDeleteOwnerRelBySchemaIds(schemaIds)), + OwnerMetaMapper.class, + mapper -> mapper.softDeleteOwnerRelBySchemaIds(schemaIds.get())), () -> SessionUtils.doWithoutCommit( SecurableObjectMapper.class, - mapper -> mapper.softDeleteObjectRelsBySchemaIds(schemaIds)), + mapper -> mapper.softDeleteObjectRelsBySchemaIds(schemaIds.get())), () -> SessionUtils.doWithoutCommit( TagMetadataObjectRelMapper.class, - mapper -> mapper.softDeleteTagMetadataObjectRelsBySchemaIds(schemaIds)), + mapper -> mapper.softDeleteTagMetadataObjectRelsBySchemaIds(schemaIds.get())), () -> SessionUtils.doWithoutCommit( PolicyMetadataObjectRelMapper.class, - mapper -> mapper.softDeletePolicyMetadataObjectRelsBySchemaIds(schemaIds)), + mapper -> mapper.softDeletePolicyMetadataObjectRelsBySchemaIds(schemaIds.get())), () -> SessionUtils.doWithoutCommit( ModelVersionAliasRelMapper.class, - mapper -> mapper.softDeleteModelVersionAliasRelsBySchemaIds(schemaIds)), + mapper -> mapper.softDeleteModelVersionAliasRelsBySchemaIds(schemaIds.get())), () -> SessionUtils.doWithoutCommit( ModelVersionMetaMapper.class, - mapper -> mapper.softDeleteModelVersionMetasBySchemaIds(schemaIds)), + mapper -> mapper.softDeleteModelVersionMetasBySchemaIds(schemaIds.get())), () -> SessionUtils.doWithoutCommit( ModelMetaMapper.class, - mapper -> mapper.softDeleteModelMetasBySchemaIds(schemaIds)), + mapper -> mapper.softDeleteModelMetasBySchemaIds(schemaIds.get())), () -> SessionUtils.doWithoutCommit( StatisticMetaMapper.class, - mapper -> mapper.softDeleteStatisticsBySchemaIds(schemaIds)), + mapper -> mapper.softDeleteStatisticsBySchemaIds(schemaIds.get())), () -> SessionUtils.doWithoutCommit( ViewMetaMapper.class, - mapper -> mapper.softDeleteViewMetasBySchemaIds(schemaIds))); + mapper -> mapper.softDeleteViewMetasBySchemaIds(schemaIds.get()))); } else { - List tableEntities = - TableMetaService.getInstance() - .listTablesByNamespace( - NamespaceUtil.ofTable( - identifier.namespace().level(0), - identifier.namespace().level(1), - schemaName)); - if (!tableEntities.isEmpty()) { - throw new NonEmptyEntityException( - "Entity %s has sub-entities, you should remove sub-entities first", identifier); - } - List filesetEntities = - FilesetMetaService.getInstance() - .listFilesetsByNamespace( - NamespaceUtil.ofFileset( - identifier.namespace().level(0), - identifier.namespace().level(1), - schemaName)); - if (!filesetEntities.isEmpty()) { - throw new NonEmptyEntityException( - "Entity %s has sub-entities, you should remove sub-entities first", identifier); - } - List modelEntities = - ModelMetaService.getInstance() - .listModelsByNamespace( - NamespaceUtil.ofModel( - identifier.namespace().level(0), - identifier.namespace().level(1), - schemaName)); - if (!modelEntities.isEmpty()) { - throw new NonEmptyEntityException( - "Entity %s has sub-entities, you should remove sub-entities first", identifier); - } - - List topicEntities = - TopicMetaService.getInstance() - .listTopicsByNamespace( - NamespaceUtil.ofTopic( - identifier.namespace().level(0), - identifier.namespace().level(1), - schemaName)); - if (!topicEntities.isEmpty()) { - throw new NonEmptyEntityException( - "Entity %s has sub-entities, you should remove sub-entities first", identifier); - } - - List singleSchemaId = Collections.singletonList(schemaId); SessionUtils.doMultipleWithCommit( - () -> - SessionUtils.doWithoutCommit( - SchemaMetaMapper.class, - mapper -> mapper.softDeleteSchemaMetasBySchemaIds(singleSchemaId)), + () -> { + // Delete the schema first and check that it was empty afterwards. The order matters: + // the delete locks the schema row, and every child write locks that same row first, so + // a table or view being created either lands before this delete and shows up in the + // check, or it waits for this transaction. Checking first would leave a gap for a child + // to appear in between. A non-empty result throws, which rolls the delete back. + lockCatalogForSchemaDelete(identifier, schemaPO); + deleteSchemaWithVersion(identifier, schemaPO); + checkSchemaIsEmpty(identifier, schemaPO); + }, () -> SessionUtils.doWithoutCommit( OwnerMetaMapper.class, @@ -416,6 +399,22 @@ public boolean deleteSchema(NameIdentifier identifier, boolean cascade) { return true; } + /** + * Soft-deletes the schema only while it still carries the version the caller read. A drop that + * lost the race must not delete a schema it never looked at. + */ + private void deleteSchemaWithVersion(NameIdentifier identifier, SchemaPO observedSchemaPO) { + int deleted = + SessionUtils.getWithoutCommit( + SchemaMetaMapper.class, + mapper -> + mapper.softDeleteSchemaMetaBySchemaIdAndVersion( + observedSchemaPO.getSchemaId(), observedSchemaPO.getCurrentVersion())); + if (deleted == 0) { + throw schemaWriteFailure(identifier, observedSchemaPO); + } + } + @Monitored( metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME, baseMetricName = "deleteSchemaMetasByLegacyTimeline") @@ -450,12 +449,196 @@ private List 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. + * + *

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. + * + *

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 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 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(); + 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) { + throw new NonEmptyEntityException( + "Entity %s has sub-entities, you should remove sub-entities first", identifier); + } + } + + /** + * Collects every HierarchicalSchema descendant of the target schema. The {@link SchemaPO} arrives + * in logical form (e.g. {@code A:B}); {@link HierarchicalConversionPOStorageOps} translates to + * storage form before running the SQL prefix match. + */ + private List listDescendantSchemaPOs(SchemaPO schemaPO) { List matched = SessionUtils.getWithoutCommit( SchemaMetaMapper.class, @@ -464,16 +647,15 @@ private List listSchemaIdsForCascade(SchemaPO schemaPO) { if (matched == null || matched.isEmpty()) { return Collections.emptyList(); } - return matched.stream().map(SchemaPO::getSchemaId).collect(Collectors.toList()); + return matched.stream() + .filter(po -> !po.getSchemaId().equals(schemaPO.getSchemaId())) + .collect(Collectors.toList()); } - private void fillSchemaPOBuilderParentEntityId(SchemaPO.Builder builder, Namespace namespace) { - NamespaceUtil.checkSchema(namespace); - NamespacedEntityId namespacedEntityId = - EntityIdService.getEntityIds( - NameIdentifier.of(namespace.levels()), Entity.EntityType.CATALOG); - builder.withMetalakeId(namespacedEntityId.namespaceIds()[0]); - builder.withCatalogId(namespacedEntityId.entityId()); + private SchemaPO.Builder newSchemaPOBuilder(CatalogPO catalogPO) { + return SchemaPO.builder() + .withMetalakeId(catalogPO.getMetalakeId()) + .withCatalogId(catalogPO.getCatalogId()); } @Monitored( 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 15a61bdd9fd..741a210e10d 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 @@ -126,6 +126,15 @@ public void insertTable(TableEntity tableEntity, boolean overwrite) throws IOExc AtomicReference tablePORef = new AtomicReference<>(); TablePO po = POConverters.initializeTablePOWithVersion(tableEntity, builder); 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, @@ -194,6 +203,18 @@ public TableEntity updateTable( final AtomicInteger updateResult = new AtomicInteger(0); try { SessionUtils.doMultipleWithCommit( + () -> { + // Only a rename that moves the table to another schema needs a lock here, and it is the + // new parent that has to stay alive, not the old one. + if (isSchemaChanged) { + SchemaMetaService.getInstance() + .lockSchemaForEntityWrite( + newTableEntity.nameIdentifier(), + newSchemaId, + oldTablePO.getCatalogId(), + oldTablePO.getMetalakeId()); + } + }, () -> updateResult.set( SessionUtils.getWithoutCommit( 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 d5618842d0e..ca33b4fe2e7 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 @@ -70,17 +70,28 @@ public void insertTopic(TopicEntity topicEntity, boolean overwrite) throws IOExc TopicPO.Builder builder = TopicPO.builder(); fillTopicPOBuilderParentEntityId(builder, topicEntity.namespace()); + TopicPO po = POConverters.initializeTopicPOWithVersion(topicEntity, builder); - SessionUtils.doWithCommit( - TopicMetaMapper.class, - mapper -> { - TopicPO po = POConverters.initializeTopicPOWithVersion(topicEntity, builder); - if (overwrite) { - mapper.insertTopicMetaOnDuplicateKeyUpdate(po); - } else { - mapper.insertTopicMeta(po); - } - }); + 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); + } + })); // TODO: insert topic dataLayout version after supporting it } catch (RuntimeException re) { ExceptionUtils.checkSQLException( 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 d676b6bf08a..50ea6f72f07 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 @@ -106,6 +106,15 @@ public void insertView(ViewEntity viewEntity, boolean overwrite) throws IOExcept 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)), diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/utils/POConverters.java b/core/src/main/java/org/apache/gravitino/storage/relational/utils/POConverters.java index 3085c1ed6d3..e166a9f4c6c 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/utils/POConverters.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/utils/POConverters.java @@ -137,8 +137,10 @@ public static MetalakePO initializeMetalakePOWithVersion(BaseMetalake baseMetala */ public static MetalakePO updateMetalakePOWithVersion( MetalakePO oldMetalakePO, BaseMetalake newMetalake) { - // Every metadata update advances the OCC token. Both version columns stay aligned because - // metalakes do not retain independently addressable historical versions. + // Every update moves the version forward, even when nothing else changes. The version is what + // the UPDATE compares against, so a version that stands still would let two servers overwrite + // each other. Both columns get the same value because a metalake keeps no old versions to + // address, unlike a fileset. Long nextVersion = oldMetalakePO.getCurrentVersion() + 1; try { return MetalakePO.builder() @@ -332,9 +334,11 @@ public static SchemaPO initializeSchemaPOWithVersion( * @return SchemaPO object with updated version */ public static SchemaPO updateSchemaPOWithVersion(SchemaPO oldSchemaPO, SchemaEntity newSchema) { - Long lastVersion = oldSchemaPO.getLastVersion(); - // Will set the version to the last version + 1 when having some fields need be multiple version - Long nextVersion = lastVersion; + // Every update moves the version forward, even when nothing else changes. The version is what + // the UPDATE compares against, so a version that stands still would let two servers overwrite + // each other. Both columns get the same value because a schema keeps no old versions to + // address, unlike a fileset. + Long nextVersion = oldSchemaPO.getCurrentVersion() + 1; try { return SchemaPO.builder() .withSchemaId(oldSchemaPO.getSchemaId()) diff --git a/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/TestSchemaMetaPostgreSQLProvider.java b/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/TestSchemaMetaPostgreSQLProvider.java new file mode 100644 index 00000000000..6c012a7bf32 --- /dev/null +++ b/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/TestSchemaMetaPostgreSQLProvider.java @@ -0,0 +1,64 @@ +/* + * 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.postgresql; + +import java.util.Collections; +import org.apache.gravitino.storage.relational.mapper.SchemaMetaMapper; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +public class TestSchemaMetaPostgreSQLProvider { + + private static final SchemaMetaPostgreSQLProvider PROVIDER = new SchemaMetaPostgreSQLProvider(); + + @Test + void testOverwriteInsertQualifiesVersionColumns() { + assertQualifiedAndAdvanced(conflictClause(PROVIDER.insertSchemaMetaOnDuplicateKeyUpdate(null))); + } + + @Test + void testBatchOverwriteInsertQualifiesVersionColumns() { + assertQualifiedAndAdvanced( + conflictClause( + PROVIDER.batchInsertSchemaMetaOnDuplicateKeyUpdate(Collections.emptyList()))); + } + + private void assertQualifiedAndAdvanced(String conflictClause) { + // PostgreSQL rejects a bare column name on this side of ON CONFLICT, because it could mean + // either the stored row or the rejected one. Both assignments must name the table. + Assertions.assertFalse( + conflictClause.matches(".*[^.\\w]current_version\\s*\\+.*"), + () -> "Found an unqualified current_version reference in: " + conflictClause); + + // An overwrite must never write the initial version back, or a stale writer could still pass + // its own version check afterwards. + Assertions.assertTrue( + conflictClause.contains( + "current_version = " + SchemaMetaMapper.TABLE_NAME + ".current_version + 1"), + () -> "current_version must advance in: " + conflictClause); + Assertions.assertTrue( + conflictClause.contains( + "last_version = " + SchemaMetaMapper.TABLE_NAME + ".current_version + 1"), + () -> "last_version must advance in: " + conflictClause); + } + + private String conflictClause(String sql) { + return sql.substring(sql.indexOf("ON CONFLICT")); + } +} diff --git a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestFunctionMetaService.java b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestFunctionMetaService.java index fe5d46db8b0..9cc3e039d0c 100644 --- a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestFunctionMetaService.java +++ b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestFunctionMetaService.java @@ -44,6 +44,7 @@ import org.apache.gravitino.authorization.SecurableObject; import org.apache.gravitino.authorization.SecurableObjects; import org.apache.gravitino.exceptions.NoSuchEntityException; +import org.apache.gravitino.exceptions.OptimisticLockException; import org.apache.gravitino.integration.test.util.GravitinoITUtils; import org.apache.gravitino.meta.FunctionEntity; import org.apache.gravitino.meta.RoleEntity; @@ -241,6 +242,72 @@ public void testUpdateFunction() throws IOException { assertTrue(versions.containsKey(2)); } + @TestTemplate + public void testUpdateFunctionFailsWhenSchemaIsDeletedConcurrently() throws IOException { + String functionName = GravitinoITUtils.genRandomName("test_function"); + Namespace namespace = NamespaceUtil.ofFunction(metalakeName, catalogName, schemaName); + FunctionEntity function = + createFunctionEntity( + RandomIdGenerator.INSTANCE.nextId(), namespace, functionName, AUDIT_INFO); + FunctionMetaService.getInstance().insertFunction(function, false); + + NameIdentifier functionIdent = + NameIdentifier.of(metalakeName, catalogName, schemaName, functionName); + NameIdentifier schemaIdent = NameIdentifier.of(metalakeName, catalogName, schemaName); + FunctionEntity updatedFunction = copyFunctionWithComment(function, "updated comment"); + + assertThrows( + NoSuchEntityException.class, + () -> + FunctionMetaService.getInstance() + .updateFunction( + functionIdent, + ignored -> { + // Reproduce the exact race deterministically: the update has already read the + // function, then the schema cascade commits before the write transaction. + assertTrue(SchemaMetaService.getInstance().deleteSchema(schemaIdent, true)); + return updatedFunction; + })); + + Map versions = listFunctionVersions(function.id()); + assertEquals(1, versions.size()); + assertVersionSoftDeleted(versions, 1); + assertFalse(versions.containsKey(2)); + } + + @TestTemplate + public void testUpdateFunctionRollsBackNewVersionAfterConcurrentDelete() throws IOException { + String functionName = GravitinoITUtils.genRandomName("test_function"); + Namespace namespace = NamespaceUtil.ofFunction(metalakeName, catalogName, schemaName); + FunctionEntity function = + createFunctionEntity( + RandomIdGenerator.INSTANCE.nextId(), namespace, functionName, AUDIT_INFO); + FunctionMetaService.getInstance().insertFunction(function, false); + + NameIdentifier functionIdent = + NameIdentifier.of(metalakeName, catalogName, schemaName, functionName); + FunctionEntity updatedFunction = copyFunctionWithComment(function, "updated comment"); + + assertThrows( + OptimisticLockException.class, + () -> + FunctionMetaService.getInstance() + .updateFunction( + functionIdent, + ignored -> { + // Delete only the function so the parent-schema lock still succeeds. The + // compare-and-set below must notice the missing function and roll version 2 + // back with the transaction. + assertTrue(FunctionMetaService.getInstance().deleteFunction(functionIdent)); + return updatedFunction; + })); + + Map versions = listFunctionVersions(function.id()); + assertEquals(1, versions.size()); + assertVersionSoftDeleted(versions, 1); + assertFalse(versions.containsKey(2)); + } + @TestTemplate public void testDeleteFunction() throws IOException { String functionName = GravitinoITUtils.genRandomName("test_function"); @@ -630,6 +697,19 @@ private Map listFunctionVersions(Long functionId) { return versionDeletedTime; } + private FunctionEntity copyFunctionWithComment(FunctionEntity function, String comment) { + return FunctionEntity.builder() + .withId(function.id()) + .withName(function.name()) + .withNamespace(function.namespace()) + .withComment(comment) + .withFunctionType(function.functionType()) + .withDeterministic(function.deterministic()) + .withDefinitions(function.definitions()) + .withAuditInfo(function.auditInfo()) + .build(); + } + private void assertVersionActive(Map versionDeletedMap, int version) { assertTrue(versionDeletedMap.containsKey(version)); assertEquals(0L, versionDeletedMap.get(version)); diff --git a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestMetalakeMetaService.java b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestMetalakeMetaService.java index 7948c75a9cf..952416c1cf5 100644 --- a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestMetalakeMetaService.java +++ b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestMetalakeMetaService.java @@ -25,21 +25,33 @@ import java.io.IOException; import java.time.Instant; import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import org.apache.gravitino.Entity; import org.apache.gravitino.EntityAlreadyExistsException; import org.apache.gravitino.exceptions.NoSuchEntityException; import org.apache.gravitino.exceptions.NonEmptyEntityException; import org.apache.gravitino.exceptions.OptimisticLockException; import org.apache.gravitino.meta.BaseMetalake; +import org.apache.gravitino.meta.CatalogEntity; +import org.apache.gravitino.meta.SchemaEntity; import org.apache.gravitino.meta.SchemaVersion; import org.apache.gravitino.storage.RandomIdGenerator; import org.apache.gravitino.storage.relational.TestJDBCBackend; import org.apache.gravitino.storage.relational.mapper.MetalakeMetaMapper; +import org.apache.gravitino.storage.relational.mapper.SchemaMetaMapper; import org.apache.gravitino.storage.relational.po.MetalakePO; +import org.apache.gravitino.storage.relational.po.SchemaPO; import org.apache.gravitino.storage.relational.utils.POConverters; import org.apache.gravitino.storage.relational.utils.SessionUtils; +import org.apache.gravitino.utils.NamespaceUtil; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.TestTemplate; +import org.mockito.Mockito; public class TestMetalakeMetaService extends TestJDBCBackend { @@ -246,6 +258,144 @@ public void testDeleteReportsOptimisticLockConflict() throws IOException { assertTrue(backend.exists(metalake.nameIdentifier(), Entity.EntityType.METALAKE)); } + @TestTemplate + public void testCascadeDeleteReportsConcurrentSchemaAlter() throws Exception { + BaseMetalake metalake = createAndInsertMakeLake(METALAKE_NAME); + CatalogEntity catalog = createAndInsertCatalog(METALAKE_NAME, "catalog"); + SchemaEntity schema = + createSchemaEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofSchema(METALAKE_NAME, catalog.name()), + "schema", + AUDIT_INFO); + backend.insert(schema, false); + SchemaPO schemaBeforeDelete = + SessionUtils.getWithoutCommit( + SchemaMetaMapper.class, mapper -> mapper.selectSchemaMetaById(schema.id())); + + ExecutorService executor = Executors.newSingleThreadExecutor(); + MetalakeMetaService service = Mockito.spy(MetalakeMetaService.getInstance()); + try { + Mockito.doAnswer( + invocation -> { + Assertions.assertEquals(metalake.id(), invocation.getArgument(0)); + List schemaPOs = + SessionUtils.getWithoutCommit( + SchemaMetaMapper.class, + mapper -> mapper.listSchemaPOsByMetalakeId(metalake.id())); + SchemaPO observedSchemaPO = + schemaPOs.stream() + .filter(schemaPO -> schemaPO.getSchemaId().equals(schema.id())) + .findFirst() + .orElseThrow(); + SchemaEntity competingSchema = + SchemaEntity.builder() + .withId(schema.id()) + .withName(schema.name()) + .withNamespace(schema.namespace()) + .withComment("competing update") + .withProperties(schema.properties()) + .withAuditInfo(schema.auditInfo()) + .build(); + SchemaPO competingSchemaPO = + POConverters.updateSchemaPOWithVersion(observedSchemaPO, competingSchema); + Future competingUpdate = + executor.submit( + () -> + SessionUtils.doWithCommitAndFetchResult( + SchemaMetaMapper.class, + mapper -> + mapper.updateSchemaMeta(competingSchemaPO, observedSchemaPO))); + Assertions.assertEquals(1, competingUpdate.get(30, TimeUnit.SECONDS)); + return schemaPOs; + }) + .when(service) + .listSchemaPOsForCascade(metalake.id()); + + assertThrows( + OptimisticLockException.class, + () -> service.deleteMetalake(metalake.nameIdentifier(), true)); + } finally { + executor.shutdownNow(); + } + + assertTrue(backend.exists(metalake.nameIdentifier(), Entity.EntityType.METALAKE)); + assertTrue(backend.exists(catalog.nameIdentifier(), Entity.EntityType.CATALOG)); + assertTrue(backend.exists(schema.nameIdentifier(), Entity.EntityType.SCHEMA)); + SchemaPO schemaAfterDelete = + SessionUtils.getWithoutCommit( + SchemaMetaMapper.class, mapper -> mapper.selectSchemaMetaById(schema.id())); + Assertions.assertEquals( + schemaBeforeDelete.getCurrentVersion() + 1, schemaAfterDelete.getCurrentVersion()); + Assertions.assertEquals("competing update", schemaAfterDelete.getSchemaComment()); + } + + @TestTemplate + public void testConcurrentMetalakeCascadeAndSchemaCreateLeavesNoOrphan() throws Exception { + BaseMetalake metalake = createAndInsertMakeLake(METALAKE_NAME); + CatalogEntity catalog = createAndInsertCatalog(METALAKE_NAME, "catalog"); + SchemaEntity schema = + createSchemaEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofSchema(METALAKE_NAME, catalog.name()), + "concurrent_schema", + AUDIT_INFO); + CountDownLatch catalogsLocked = new CountDownLatch(1); + CountDownLatch allowSchemaSnapshot = new CountDownLatch(1); + CountDownLatch createStarted = new CountDownLatch(1); + ExecutorService executor = Executors.newFixedThreadPool(2); + MetalakeMetaService service = Mockito.spy(MetalakeMetaService.getInstance()); + try { + Mockito.doAnswer( + invocation -> { + catalogsLocked.countDown(); + assertTrue(allowSchemaSnapshot.await(30, TimeUnit.SECONDS)); + return invocation.callRealMethod(); + }) + .when(service) + .listSchemaPOsForCascade(metalake.id()); + + Future deleteResult = + executor.submit( + () -> { + try { + service.deleteMetalake(metalake.nameIdentifier(), true); + return null; + } catch (Throwable throwable) { + return throwable; + } + }); + assertTrue(catalogsLocked.await(30, TimeUnit.SECONDS)); + + Future createResult = + executor.submit( + () -> { + createStarted.countDown(); + try { + SchemaMetaService.getInstance().insertSchema(schema, false); + return null; + } catch (Throwable throwable) { + return throwable; + } + }); + assertTrue(createStarted.await(30, TimeUnit.SECONDS)); + assertThrows(TimeoutException.class, () -> createResult.get(500, TimeUnit.MILLISECONDS)); + + allowSchemaSnapshot.countDown(); + Throwable createFailure = createResult.get(30, TimeUnit.SECONDS); + Throwable deleteFailure = deleteResult.get(30, TimeUnit.SECONDS); + Assertions.assertInstanceOf(NoSuchEntityException.class, createFailure); + Assertions.assertNull(deleteFailure, () -> "Metalake cascade failed: " + deleteFailure); + } finally { + allowSchemaSnapshot.countDown(); + executor.shutdownNow(); + } + + assertFalse(backend.exists(metalake.nameIdentifier(), Entity.EntityType.METALAKE)); + assertFalse(backend.exists(catalog.nameIdentifier(), Entity.EntityType.CATALOG)); + assertFalse(backend.exists(schema.nameIdentifier(), Entity.EntityType.SCHEMA)); + } + @TestTemplate public void testNonCascadeDeleteRollsBackMetalakeFence() throws IOException { BaseMetalake metalake = createAndInsertMakeLake(METALAKE_NAME); diff --git a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestModelVersionMetaService.java b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestModelVersionMetaService.java index bc37a31cf1a..888c577b4ed 100644 --- a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestModelVersionMetaService.java +++ b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestModelVersionMetaService.java @@ -28,6 +28,12 @@ import java.util.List; import java.util.Map; import java.util.UUID; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.function.Function; import java.util.stream.Collectors; import org.apache.gravitino.Entity; @@ -39,9 +45,15 @@ import org.apache.gravitino.meta.AuditInfo; import org.apache.gravitino.meta.ModelEntity; import org.apache.gravitino.meta.ModelVersionEntity; +import org.apache.gravitino.meta.SchemaEntity; import org.apache.gravitino.model.ModelVersion; import org.apache.gravitino.storage.RandomIdGenerator; import org.apache.gravitino.storage.relational.TestJDBCBackend; +import org.apache.gravitino.storage.relational.mapper.ModelVersionAliasRelMapper; +import org.apache.gravitino.storage.relational.mapper.ModelVersionMetaMapper; +import org.apache.gravitino.storage.relational.mapper.SchemaMetaMapper; +import org.apache.gravitino.storage.relational.po.SchemaPO; +import org.apache.gravitino.storage.relational.utils.SessionUtils; import org.apache.gravitino.utils.NameIdentifierUtil; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.TestTemplate; @@ -60,6 +72,162 @@ public class TestModelVersionMetaService extends TestJDBCBackend { private final List aliases = Lists.newArrayList("alias1", "alias2"); + @TestTemplate + public void testInsertModelVersionWaitsForConcurrentSchemaDelete() throws Exception { + createParentEntities(METALAKE_NAME, CATALOG_NAME, SCHEMA_NAME, AUDIT_INFO); + SchemaEntity schema = + SchemaMetaService.getInstance() + .getSchemaByIdentifier(NameIdentifier.of(METALAKE_NAME, CATALOG_NAME, SCHEMA_NAME)); + SchemaPO observedSchemaPO = + SessionUtils.getWithoutCommit( + SchemaMetaMapper.class, mapper -> mapper.selectSchemaMetaById(schema.id())); + + ModelEntity modelEntity = + createModelEntity( + RandomIdGenerator.INSTANCE.nextId(), + MODEL_NS, + "model_racing_schema_drop", + "model comment", + 0, + properties, + AUDIT_INFO); + ModelMetaService.getInstance().insertModel(modelEntity, false); + ModelVersionEntity modelVersionEntity = + createModelVersionEntity( + modelEntity.nameIdentifier(), + 0, + ImmutableMap.of(ModelVersion.URI_NAME_UNKNOWN, "model_path"), + aliases, + "version comment", + properties, + AUDIT_INFO); + + CountDownLatch schemaDeleteLocked = new CountDownLatch(1); + CountDownLatch allowDeleteCommit = new CountDownLatch(1); + CountDownLatch versionInsertStarted = new CountDownLatch(1); + ExecutorService executor = Executors.newFixedThreadPool(2); + Future 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 { + Assertions.assertTrue(allowDeleteCommit.await(30, TimeUnit.SECONDS)); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException(e); + } + }); + return null; + } catch (Throwable throwable) { + return throwable; + } + }); + try { + Assertions.assertTrue(schemaDeleteLocked.await(30, TimeUnit.SECONDS)); + Future insertResult = + executor.submit( + () -> { + versionInsertStarted.countDown(); + try { + ModelVersionMetaService.getInstance().insertModelVersion(modelVersionEntity); + return null; + } catch (Throwable throwable) { + return throwable; + } + }); + Assertions.assertTrue(versionInsertStarted.await(30, TimeUnit.SECONDS)); + Assertions.assertThrows( + TimeoutException.class, () -> insertResult.get(500, TimeUnit.MILLISECONDS)); + + allowDeleteCommit.countDown(); + Assertions.assertNull(deleteResult.get(30, TimeUnit.SECONDS)); + Assertions.assertInstanceOf( + NoSuchEntityException.class, insertResult.get(30, TimeUnit.SECONDS)); + } finally { + allowDeleteCommit.countDown(); + executor.shutdownNow(); + } + + Assertions.assertTrue( + SessionUtils.getWithoutCommit( + ModelVersionMetaMapper.class, + mapper -> mapper.listModelVersionMetasByModelId(modelEntity.id())) + .isEmpty()); + Assertions.assertTrue( + SessionUtils.getWithoutCommit( + ModelVersionAliasRelMapper.class, + mapper -> mapper.selectModelVersionAliasRelsByModelId(modelEntity.id())) + .isEmpty()); + } + + @TestTemplate + public void testUpdateModelVersionFailsWhenSchemaIsDeletedConcurrently() throws IOException { + createParentEntities(METALAKE_NAME, CATALOG_NAME, SCHEMA_NAME, AUDIT_INFO); + ModelEntity model = + createModelEntity( + RandomIdGenerator.INSTANCE.nextId(), + MODEL_NS, + "model_updated_during_schema_drop", + "model comment", + 0, + properties, + AUDIT_INFO); + ModelMetaService.getInstance().insertModel(model, false); + ModelVersionEntity modelVersion = + createModelVersionEntity( + model.nameIdentifier(), + 0, + ImmutableMap.of(ModelVersion.URI_NAME_UNKNOWN, "old_path"), + aliases, + "version comment", + properties, + AUDIT_INFO); + ModelVersionMetaService.getInstance().insertModelVersion(modelVersion); + + ModelVersionEntity updatedVersion = + createModelVersionEntity( + model.nameIdentifier(), + 0, + ImmutableMap.of(ModelVersion.URI_NAME_UNKNOWN, "new_path"), + aliases, + "version comment", + properties, + AUDIT_INFO); + NameIdentifier schemaIdent = NameIdentifier.of(METALAKE_NAME, CATALOG_NAME, SCHEMA_NAME); + + Assertions.assertThrows( + NoSuchEntityException.class, + () -> + ModelVersionMetaService.getInstance() + .updateModelVersion( + modelVersion.nameIdentifier(), + ignored -> { + // The update has already resolved both the model and its version here. A + // schema cascade that commits now must make the write fail before it can + // reinsert the version with its new URI. + Assertions.assertTrue( + SchemaMetaService.getInstance().deleteSchema(schemaIdent, true)); + return updatedVersion; + })); + + Assertions.assertTrue( + SessionUtils.getWithoutCommit( + ModelVersionMetaMapper.class, + mapper -> mapper.listModelVersionMetasByModelId(model.id())) + .isEmpty()); + } + @TestTemplate public void testInsertAndSelectModelVersion() throws IOException { createParentEntities(METALAKE_NAME, CATALOG_NAME, SCHEMA_NAME, AUDIT_INFO); 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 a1e43144ec8..72d721a4c28 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 @@ -28,15 +28,26 @@ import java.sql.SQLException; import java.sql.Statement; import java.time.Instant; +import java.util.Arrays; import java.util.Collections; import java.util.List; +import java.util.Objects; import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.stream.Collectors; import org.apache.gravitino.Entity; import org.apache.gravitino.EntityAlreadyExistsException; import org.apache.gravitino.NameIdentifier; import org.apache.gravitino.Namespace; +import org.apache.gravitino.exceptions.NoSuchEntityException; import org.apache.gravitino.exceptions.NonEmptyEntityException; +import org.apache.gravitino.exceptions.OptimisticLockException; +import org.apache.gravitino.meta.CatalogEntity; import org.apache.gravitino.meta.ColumnEntity; import org.apache.gravitino.meta.FilesetEntity; import org.apache.gravitino.meta.FunctionEntity; @@ -49,12 +60,19 @@ import org.apache.gravitino.rel.types.Types; import org.apache.gravitino.storage.RandomIdGenerator; import org.apache.gravitino.storage.relational.TestJDBCBackend; +import org.apache.gravitino.storage.relational.mapper.CatalogMetaMapper; +import org.apache.gravitino.storage.relational.mapper.SchemaMetaMapper; +import org.apache.gravitino.storage.relational.po.CatalogPO; +import org.apache.gravitino.storage.relational.po.SchemaPO; import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper; +import org.apache.gravitino.storage.relational.utils.POConverters; +import org.apache.gravitino.storage.relational.utils.SessionUtils; import org.apache.gravitino.utils.NameIdentifierUtil; import org.apache.gravitino.utils.NamespaceUtil; import org.apache.ibatis.session.SqlSession; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.TestTemplate; +import org.mockito.Mockito; public class TestSchemaMetaService extends TestJDBCBackend { private final String metalakeName = "metalake_for_catalog_test"; @@ -81,6 +99,237 @@ public void testInsertAlreadyExistsException() throws IOException { assertThrows(EntityAlreadyExistsException.class, () -> backend.insert(schemaCopy, false)); } + @TestTemplate + public void testInsertSchemaLocksCatalogWithoutChangingVersion() throws IOException { + createAndInsertMakeLake(metalakeName); + CatalogEntity catalog = createAndInsertCatalog(metalakeName, catalogName); + CatalogPO beforeInsert = + SessionUtils.getWithoutCommit( + CatalogMetaMapper.class, mapper -> mapper.selectCatalogMetaById(catalog.id())); + SchemaEntity schema = + createSchemaEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofSchema(metalakeName, catalogName), + "schema_fence", + AUDIT_INFO); + backend.insert(schema, false); + + CatalogPO afterInsert = + SessionUtils.getWithoutCommit( + CatalogMetaMapper.class, mapper -> mapper.selectCatalogMetaById(catalog.id())); + Assertions.assertEquals(beforeInsert.getCurrentVersion(), afterInsert.getCurrentVersion()); + Assertions.assertEquals(beforeInsert.getLastVersion(), afterInsert.getLastVersion()); + + SchemaEntity duplicate = + createSchemaEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofSchema(metalakeName, catalogName), + schema.name(), + AUDIT_INFO); + assertThrows(EntityAlreadyExistsException.class, () -> backend.insert(duplicate, false)); + + CatalogPO afterFailure = + SessionUtils.getWithoutCommit( + CatalogMetaMapper.class, mapper -> mapper.selectCatalogMetaById(catalog.id())); + Assertions.assertEquals(afterInsert.getCurrentVersion(), afterFailure.getCurrentVersion()); + Assertions.assertEquals(afterInsert.getLastVersion(), afterFailure.getLastVersion()); + } + + @TestTemplate + public void testSchemaChildServicesWaitForConcurrentSchemaDelete() throws Exception { + 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)); + + for (int index = 0; index < childWrites.size(); index++) { + SchemaEntity schema = + createSchemaEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofSchema(metalakeName, catalogName), + "schema_for_entity_lock_" + index, + AUDIT_INFO); + backend.insert(schema, false); + assertChildWriteWaitsForConcurrentSchemaDelete(schema, childWrites.get(index)); + } + } + + private void assertChildWriteWaitsForConcurrentSchemaDelete( + SchemaEntity schema, SchemaChildWrite childWrite) 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); + ExecutorService executor = Executors.newFixedThreadPool(2); + Future 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)); + Future createResult = + executor.submit( + () -> { + entityCreateStarted.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())); + return null; + } catch (Throwable throwable) { + return throwable; + } + }); + assertTrue(entityCreateStarted.await(30, TimeUnit.SECONDS)); + assertThrows(TimeoutException.class, () -> createResult.get(500, TimeUnit.MILLISECONDS)); + + allowDeleteCommit.countDown(); + Assertions.assertNull(deleteResult.get(30, TimeUnit.SECONDS)); + Assertions.assertInstanceOf( + NoSuchEntityException.class, createResult.get(30, TimeUnit.SECONDS)); + } finally { + allowDeleteCommit.countDown(); + executor.shutdownNow(); + } + } + + @TestTemplate + public void testConcurrentSameNameSchemaCreateReportsAlreadyExists() throws Exception { + createAndInsertMakeLake(metalakeName); + createAndInsertCatalog(metalakeName, catalogName); + SchemaEntity first = + createSchemaEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofSchema(metalakeName, catalogName), + "concurrent_schema", + AUDIT_INFO); + SchemaEntity second = + createSchemaEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofSchema(metalakeName, catalogName), + first.name(), + AUDIT_INFO); + + List results = insertSchemasConcurrently(first, second); + Assertions.assertEquals(1, results.stream().filter(Objects::isNull).count()); + Throwable failure = results.stream().filter(Objects::nonNull).findFirst().orElseThrow(); + Assertions.assertTrue( + failure instanceof EntityAlreadyExistsException, + () -> "Expected EntityAlreadyExistsException, but got " + failure); + } + + @TestTemplate + public void testConcurrentDifferentSchemaCreatesBothSucceed() throws Exception { + createAndInsertMakeLake(metalakeName); + createAndInsertCatalog(metalakeName, catalogName); + SchemaEntity first = + createSchemaEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofSchema(metalakeName, catalogName), + "concurrent_schema_1", + AUDIT_INFO); + SchemaEntity second = + createSchemaEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofSchema(metalakeName, catalogName), + "concurrent_schema_2", + AUDIT_INFO); + + List results = insertSchemasConcurrently(first, second); + Assertions.assertTrue( + results.stream().allMatch(Objects::isNull), + () -> "Concurrent schema creates failed: " + results); + } + + @TestTemplate + public void testConcurrentSameSchemaDeletesAreIdempotent() throws Exception { + createAndInsertMakeLake(metalakeName); + createAndInsertCatalog(metalakeName, catalogName); + SchemaEntity schema = + createSchemaEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofSchema(metalakeName, catalogName), + "concurrent_delete_schema", + AUDIT_INFO); + backend.insert(schema, false); + + List results = + deleteSchemasConcurrently(schema.nameIdentifier(), schema.nameIdentifier()); + Assertions.assertEquals(1, results.stream().filter(Objects::isNull).count()); + Throwable loser = results.stream().filter(Objects::nonNull).findFirst().orElseThrow(); + Assertions.assertTrue( + loser instanceof NoSuchEntityException, + () -> "Expected an idempotent missing result, but got " + loser); + } + @TestTemplate public void testUpdateAlreadyExistsException() throws IOException { createAndInsertMakeLake(metalakeName); @@ -145,6 +394,124 @@ public void testUpdateSchemaCommentFromNull() throws IOException { Assertions.assertEquals("schema comment updated", updatedSchema.comment()); } + @TestTemplate + public void testAlterAndDeleteUseCurrentVersion() throws IOException { + createAndInsertMakeLake(metalakeName); + createAndInsertCatalog(metalakeName, catalogName); + SchemaEntity schema = + createSchemaEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofSchema(metalakeName, catalogName), + "schema_occ", + AUDIT_INFO); + backend.insert(schema, false); + SchemaPO oldPO = + SessionUtils.getWithoutCommit( + SchemaMetaMapper.class, mapper -> mapper.selectSchemaMetaById(schema.id())); + SchemaEntity updatedSchema = + SchemaEntity.builder() + .withId(schema.id()) + .withName(schema.name()) + .withNamespace(schema.namespace()) + .withAuditInfo(schema.auditInfo()) + .withComment("updated") + .withProperties(schema.properties()) + .build(); + SchemaPO newPO = POConverters.updateSchemaPOWithVersion(oldPO, updatedSchema); + + int updated = + SessionUtils.doWithCommitAndFetchResult( + SchemaMetaMapper.class, mapper -> mapper.updateSchemaMeta(newPO, oldPO)); + int staleUpdate = + SessionUtils.doWithCommitAndFetchResult( + SchemaMetaMapper.class, mapper -> mapper.updateSchemaMeta(newPO, oldPO)); + int staleDelete = + SessionUtils.doWithCommitAndFetchResult( + SchemaMetaMapper.class, + mapper -> + mapper.softDeleteSchemaMetaBySchemaIdAndVersion( + schema.id(), oldPO.getCurrentVersion())); + Assertions.assertEquals(1, updated); + Assertions.assertEquals(0, staleUpdate); + Assertions.assertEquals(0, staleDelete); + assertTrue(backend.exists(schema.nameIdentifier(), Entity.EntityType.SCHEMA)); + int deleted = + SessionUtils.doWithCommitAndFetchResult( + SchemaMetaMapper.class, + mapper -> + mapper.softDeleteSchemaMetaBySchemaIdAndVersion( + schema.id(), newPO.getCurrentVersion())); + Assertions.assertEquals(1, deleted); + } + + @TestTemplate + public void testAlterReportsOptimisticLockConflict() throws IOException { + createAndInsertMakeLake(metalakeName); + createAndInsertCatalog(metalakeName, catalogName); + SchemaEntity schema = + createSchemaEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofSchema(metalakeName, catalogName), + "schema_alter_conflict", + AUDIT_INFO); + backend.insert(schema, false); + + assertThrows( + OptimisticLockException.class, + () -> + SchemaMetaService.getInstance() + .updateSchema( + schema.nameIdentifier(), + entity -> { + SchemaEntity current = (SchemaEntity) entity; + SchemaPO currentPO = + SessionUtils.getWithoutCommit( + SchemaMetaMapper.class, + mapper -> mapper.selectSchemaMetaById(current.id())); + SchemaEntity competingUpdate = + copySchemaWithComment(current, "competing update"); + SchemaPO competingPO = + POConverters.updateSchemaPOWithVersion(currentPO, competingUpdate); + SessionUtils.doWithCommitAndFetchResult( + SchemaMetaMapper.class, + mapper -> mapper.updateSchemaMeta(competingPO, currentPO)); + return copySchemaWithComment(current, "requested update"); + })); + } + + @TestTemplate + public void testAlterReportsNoSuchWhenSchemaIsDeletedConcurrently() throws IOException { + createAndInsertMakeLake(metalakeName); + createAndInsertCatalog(metalakeName, catalogName); + SchemaEntity schema = + createSchemaEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofSchema(metalakeName, catalogName), + "schema_alter_deleted", + AUDIT_INFO); + backend.insert(schema, false); + + assertThrows( + NoSuchEntityException.class, + () -> + SchemaMetaService.getInstance() + .updateSchema( + schema.nameIdentifier(), + entity -> { + SchemaEntity current = (SchemaEntity) entity; + SchemaPO currentPO = + SessionUtils.getWithoutCommit( + SchemaMetaMapper.class, + mapper -> mapper.selectSchemaMetaById(current.id())); + SessionUtils.doWithCommitAndFetchResult( + SchemaMetaMapper.class, + mapper -> + mapper.softDeleteSchemaMetaBySchemaIdAndVersion( + current.id(), currentPO.getCurrentVersion())); + return copySchemaWithComment(current, "requested update"); + })); + } + @TestTemplate public void testMetaLifeCycleFromCreationToDeletion() throws IOException { createAndInsertMakeLake(metalakeName); @@ -215,16 +582,77 @@ public void testDeleteSchemaNonCascadingFailsWhenTopicExists() throws IOExceptio topicName, AUDIT_INFO); topicMetaService.insertTopic(topic, false); + SchemaPO beforeDelete = + SessionUtils.getWithoutCommit( + SchemaMetaMapper.class, mapper -> mapper.selectSchemaMetaById(schema.id())); Assertions.assertThrows( NonEmptyEntityException.class, () -> schemaMetaService.deleteSchema(schema.nameIdentifier(), false), "Non-cascading delete must fail when dependent topics exist."); + SchemaPO afterDelete = + SessionUtils.getWithoutCommit( + SchemaMetaMapper.class, mapper -> mapper.selectSchemaMetaById(schema.id())); + Assertions.assertEquals(beforeDelete.getCurrentVersion(), afterDelete.getCurrentVersion()); + assertTrue(backend.exists(schema.nameIdentifier(), Entity.EntityType.SCHEMA)); + assertTrue(backend.exists(topic.nameIdentifier(), Entity.EntityType.TOPIC)); + topicMetaService.deleteTopic(topic.nameIdentifier()); schemaMetaService.deleteSchema(schema.nameIdentifier(), false); } + @TestTemplate + public void testDeleteSchemaNonCascadingFailsWhenViewExists() throws IOException { + createAndInsertMakeLake(metalakeName); + createAndInsertCatalog(metalakeName, catalogName); + SchemaEntity schema = + createSchemaEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofSchema(metalakeName, catalogName), + "schema_with_view", + AUDIT_INFO); + SchemaMetaService.getInstance().insertSchema(schema, false); + ViewEntity view = + createViewEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofView(metalakeName, catalogName, schema.name()), + "dependent_view"); + ViewMetaService.getInstance().insertView(view, false); + + assertThrows( + NonEmptyEntityException.class, + () -> SchemaMetaService.getInstance().deleteSchema(schema.nameIdentifier(), false)); + assertTrue(backend.exists(schema.nameIdentifier(), Entity.EntityType.SCHEMA)); + assertTrue(backend.exists(view.nameIdentifier(), Entity.EntityType.VIEW)); + } + + @TestTemplate + public void testDeleteSchemaNonCascadingFailsWhenFunctionExists() throws IOException { + createAndInsertMakeLake(metalakeName); + createAndInsertCatalog(metalakeName, catalogName); + SchemaEntity schema = + createSchemaEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofSchema(metalakeName, catalogName), + "schema_with_function", + AUDIT_INFO); + SchemaMetaService.getInstance().insertSchema(schema, false); + FunctionEntity function = + createFunctionEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofFunction(metalakeName, catalogName, schema.name()), + "dependent_function", + AUDIT_INFO); + FunctionMetaService.getInstance().insertFunction(function, false); + + assertThrows( + NonEmptyEntityException.class, + () -> SchemaMetaService.getInstance().deleteSchema(schema.nameIdentifier(), false)); + assertTrue(backend.exists(schema.nameIdentifier(), Entity.EntityType.SCHEMA)); + assertTrue(backend.exists(function.nameIdentifier(), Entity.EntityType.FUNCTION)); + } + @TestTemplate public void testInsertHierarchicalSchemaCreatesAncestorsAndLeaf() throws IOException { createAndInsertMakeLake(metalakeName); @@ -338,6 +766,42 @@ public void testDeleteHierarchicalSchemaCascadeRemovesDescendantsAndChildren() NameIdentifier.of(metalakeName, catalogName, "anc_a"), Entity.EntityType.SCHEMA)); } + @TestTemplate + public void testOverlappingHierarchicalSchemaDeletesDoNotDeadlock() throws Exception { + createAndInsertMakeLake(metalakeName); + createAndInsertCatalog(metalakeName, catalogName); + SchemaMetaService schemaMetaService = SchemaMetaService.getInstance(); + String firstLeaf = "overlap_a:overlap_b:leaf_c"; + String secondLeaf = "overlap_a:overlap_b:leaf_d"; + schemaMetaService.insertSchema( + createSchemaEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofSchema(metalakeName, catalogName), + firstLeaf, + AUDIT_INFO), + false); + schemaMetaService.insertSchema( + createSchemaEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofSchema(metalakeName, catalogName), + secondLeaf, + AUDIT_INFO), + false); + + NameIdentifier ancestor = NameIdentifier.of(metalakeName, catalogName, "overlap_a:overlap_b"); + NameIdentifier descendant = NameIdentifier.of(metalakeName, catalogName, firstLeaf); + List results = deleteSchemasConcurrently(ancestor, descendant); + Assertions.assertTrue( + results.stream() + .allMatch(result -> result == null || result instanceof NoSuchEntityException), + () -> "Overlapping cascade deletes produced an unexpected failure: " + results); + Assertions.assertFalse(backend.exists(ancestor, Entity.EntityType.SCHEMA)); + Assertions.assertFalse(backend.exists(descendant, Entity.EntityType.SCHEMA)); + Assertions.assertFalse( + backend.exists( + NameIdentifier.of(metalakeName, catalogName, secondLeaf), Entity.EntityType.SCHEMA)); + } + @TestTemplate public void testDeleteSchemaCascadeRemovesTagRelations() throws IOException { createAndInsertMakeLake(metalakeName); @@ -507,14 +971,24 @@ public void testInsertHierarchicalSecondLeafReusesAncestorsWithoutUpsert() throw .build(); schemaMetaService.insertSchema(first, false); - long idA = - schemaMetaService - .getSchemaByIdentifier(NameIdentifier.of(metalakeName, catalogName, ancestorA)) - .id(); - long idAB = - schemaMetaService - .getSchemaByIdentifier(NameIdentifier.of(metalakeName, catalogName, ancestorAB)) - .id(); + SchemaPO ancestorAPOBefore = + SessionUtils.getWithoutCommit( + SchemaMetaMapper.class, + mapper -> + mapper.selectSchemaMetaById( + schemaMetaService + .getSchemaByIdentifier( + NameIdentifier.of(metalakeName, catalogName, ancestorA)) + .id())); + SchemaPO ancestorABPOBefore = + SessionUtils.getWithoutCommit( + SchemaMetaMapper.class, + mapper -> + mapper.selectSchemaMetaById( + schemaMetaService + .getSchemaByIdentifier( + NameIdentifier.of(metalakeName, catalogName, ancestorAB)) + .id())); SchemaEntity second = SchemaEntity.builder() @@ -527,16 +1001,182 @@ public void testInsertHierarchicalSecondLeafReusesAncestorsWithoutUpsert() throw .build(); schemaMetaService.insertSchema(second, false); + SchemaPO ancestorAPOAfter = + SessionUtils.getWithoutCommit( + SchemaMetaMapper.class, + mapper -> + mapper.selectSchemaMetaById( + schemaMetaService + .getSchemaByIdentifier( + NameIdentifier.of(metalakeName, catalogName, ancestorA)) + .id())); + SchemaPO ancestorABPOAfter = + SessionUtils.getWithoutCommit( + SchemaMetaMapper.class, + mapper -> + mapper.selectSchemaMetaById( + schemaMetaService + .getSchemaByIdentifier( + NameIdentifier.of(metalakeName, catalogName, ancestorAB)) + .id())); + Assertions.assertEquals(ancestorAPOBefore.getSchemaId(), ancestorAPOAfter.getSchemaId()); + Assertions.assertEquals(ancestorABPOBefore.getSchemaId(), ancestorABPOAfter.getSchemaId()); Assertions.assertEquals( - idA, - schemaMetaService - .getSchemaByIdentifier(NameIdentifier.of(metalakeName, catalogName, ancestorA)) - .id()); + ancestorAPOBefore.getCurrentVersion(), ancestorAPOAfter.getCurrentVersion()); Assertions.assertEquals( - idAB, - schemaMetaService - .getSchemaByIdentifier(NameIdentifier.of(metalakeName, catalogName, ancestorAB)) - .id()); + ancestorABPOBefore.getCurrentVersion(), ancestorABPOAfter.getCurrentVersion()); + } + + @TestTemplate + public void testConcurrentCatalogCascadeAndSchemaCreateLeavesNoOrphan() throws Exception { + createAndInsertMakeLake(metalakeName); + CatalogEntity catalog = createAndInsertCatalog(metalakeName, catalogName); + SchemaEntity schema = + createSchemaEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofSchema(metalakeName, catalogName), + "schema_racing_catalog_drop", + AUDIT_INFO); + + CountDownLatch catalogLocked = new CountDownLatch(1); + CountDownLatch allowSchemaSnapshot = new CountDownLatch(1); + CountDownLatch createStarted = new CountDownLatch(1); + ExecutorService executor = Executors.newFixedThreadPool(2); + CatalogMetaService service = Mockito.spy(CatalogMetaService.getInstance()); + try { + // Pause the cascade right after it has soft-deleted the catalog row, which is the moment it + // holds that row, and before it reads the schemas to delete. + Mockito.doAnswer( + invocation -> { + catalogLocked.countDown(); + assertTrue(allowSchemaSnapshot.await(30, TimeUnit.SECONDS)); + return invocation.callRealMethod(); + }) + .when(service) + .listSchemaPOsForCascade(catalog.id()); + + Future deleteResult = + executor.submit( + () -> { + try { + service.deleteCatalog(catalog.nameIdentifier(), true); + return null; + } catch (Throwable throwable) { + return throwable; + } + }); + assertTrue(catalogLocked.await(30, TimeUnit.SECONDS)); + + Future createResult = + executor.submit( + () -> { + createStarted.countDown(); + try { + SchemaMetaService.getInstance().insertSchema(schema, false); + return null; + } catch (Throwable throwable) { + return throwable; + } + }); + assertTrue(createStarted.await(30, TimeUnit.SECONDS)); + // The create must not slip past the drop: it waits on the catalog row instead. + assertThrows(TimeoutException.class, () -> createResult.get(500, TimeUnit.MILLISECONDS)); + + allowSchemaSnapshot.countDown(); + Throwable createFailure = createResult.get(30, TimeUnit.SECONDS); + Throwable deleteFailure = deleteResult.get(30, TimeUnit.SECONDS); + Assertions.assertInstanceOf(NoSuchEntityException.class, createFailure); + Assertions.assertNull(deleteFailure, () -> "Catalog cascade failed: " + deleteFailure); + } finally { + allowSchemaSnapshot.countDown(); + executor.shutdownNow(); + } + + assertFalse(backend.exists(catalog.nameIdentifier(), Entity.EntityType.CATALOG)); + assertFalse(backend.exists(schema.nameIdentifier(), Entity.EntityType.SCHEMA)); + } + + private List insertSchemasConcurrently(SchemaEntity first, SchemaEntity second) + throws Exception { + ExecutorService executor = Executors.newFixedThreadPool(2); + CountDownLatch ready = new CountDownLatch(2); + CountDownLatch start = new CountDownLatch(1); + try { + Future firstResult = + executor.submit( + () -> { + ready.countDown(); + start.await(); + try { + SchemaMetaService.getInstance().insertSchema(first, false); + return null; + } catch (Throwable throwable) { + return throwable; + } + }); + Future secondResult = + executor.submit( + () -> { + ready.countDown(); + start.await(); + try { + SchemaMetaService.getInstance().insertSchema(second, false); + return null; + } catch (Throwable throwable) { + return throwable; + } + }); + assertTrue(ready.await(30, TimeUnit.SECONDS)); + start.countDown(); + return Arrays.asList( + firstResult.get(30, TimeUnit.SECONDS), secondResult.get(30, TimeUnit.SECONDS)); + } finally { + start.countDown(); + executor.shutdownNow(); + } + } + + private List deleteSchemasConcurrently(NameIdentifier first, NameIdentifier second) + throws Exception { + ExecutorService executor = Executors.newFixedThreadPool(2); + CountDownLatch ready = new CountDownLatch(2); + CountDownLatch start = new CountDownLatch(1); + try { + Future firstResult = + executor.submit(() -> deleteSchemaAfterStart(first, ready, start)); + Future secondResult = + executor.submit(() -> deleteSchemaAfterStart(second, ready, start)); + assertTrue(ready.await(30, TimeUnit.SECONDS)); + start.countDown(); + return Arrays.asList( + firstResult.get(30, TimeUnit.SECONDS), secondResult.get(30, TimeUnit.SECONDS)); + } finally { + start.countDown(); + executor.shutdownNow(); + } + } + + private Throwable deleteSchemaAfterStart( + NameIdentifier identifier, CountDownLatch ready, CountDownLatch start) { + ready.countDown(); + try { + start.await(); + SchemaMetaService.getInstance().deleteSchema(identifier, true); + return null; + } catch (Throwable throwable) { + return throwable; + } + } + + private SchemaEntity copySchemaWithComment(SchemaEntity schema, String comment) { + return SchemaEntity.builder() + .withId(schema.id()) + .withName(schema.name()) + .withNamespace(schema.namespace()) + .withComment(comment) + .withProperties(schema.properties()) + .withAuditInfo(schema.auditInfo()) + .build(); } private void associateTag(TagEntity tag, NameIdentifier ident, Entity.EntityType type) @@ -569,4 +1209,9 @@ private int countActiveTagRelForMetadataObject(Long metadataObjectId, String met throw new RuntimeException("SQL execution failed", e); } } + + @FunctionalInterface + private interface SchemaChildWrite { + void run(Namespace namespace) throws Exception; + } } diff --git a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestTableMetaService.java b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestTableMetaService.java index a795168a5d2..40d737bca93 100644 --- a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestTableMetaService.java +++ b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestTableMetaService.java @@ -27,6 +27,12 @@ import java.time.Instant; import java.util.List; import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.function.Function; import java.util.stream.Collectors; import org.apache.gravitino.Entity; @@ -56,6 +62,8 @@ import org.apache.gravitino.storage.RandomIdGenerator; import org.apache.gravitino.storage.relational.TestJDBCBackend; import org.apache.gravitino.storage.relational.mapper.EntityChangeLogMapper; +import org.apache.gravitino.storage.relational.mapper.SchemaMetaMapper; +import org.apache.gravitino.storage.relational.po.SchemaPO; import org.apache.gravitino.storage.relational.po.cache.EntityChangeRecord; import org.apache.gravitino.storage.relational.po.cache.OperateType; import org.apache.gravitino.storage.relational.utils.SessionUtils; @@ -318,6 +326,106 @@ record -> && record.getOperateType() == OperateType.DROP)); } + @TestTemplate + public void testMoveTableWaitsForConcurrentTargetSchemaDelete() throws Exception { + String sourceSchemaName = "source_schema"; + String targetSchemaName = "target_schema"; + createParentEntities(metalakeName, catalogName, sourceSchemaName, AUDIT_INFO); + SchemaEntity targetSchema = + createSchemaEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofSchema(metalakeName, catalogName), + targetSchemaName, + AUDIT_INFO); + backend.insert(targetSchema, false); + TableEntity table = + createTableEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofTable(metalakeName, catalogName, sourceSchemaName), + "moving_table", + AUDIT_INFO); + backend.insert(table, false); + + SchemaPO observedTargetSchema = + SessionUtils.getWithoutCommit( + SchemaMetaMapper.class, mapper -> mapper.selectSchemaMetaById(targetSchema.id())); + CountDownLatch targetDeleteLocked = new CountDownLatch(1); + CountDownLatch allowDeleteCommit = new CountDownLatch(1); + CountDownLatch moveStarted = new CountDownLatch(1); + ExecutorService executor = Executors.newFixedThreadPool(2); + Future deleteResult = + executor.submit( + () -> { + try { + SessionUtils.doMultipleWithCommit( + () -> { + int deleted = + SessionUtils.getWithoutCommit( + SchemaMetaMapper.class, + mapper -> + mapper.softDeleteSchemaMetaBySchemaIdAndVersion( + observedTargetSchema.getSchemaId(), + observedTargetSchema.getCurrentVersion())); + Assertions.assertEquals(1, deleted); + targetDeleteLocked.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(targetDeleteLocked.await(30, TimeUnit.SECONDS)); + Future moveResult = + executor.submit( + () -> { + moveStarted.countDown(); + try { + TableEntity movedTable = + TableEntity.builder() + .withId(table.id()) + .withName(table.name()) + .withNamespace( + NamespaceUtil.ofTable(metalakeName, catalogName, targetSchemaName)) + .withColumns(table.columns()) + .withAuditInfo(table.auditInfo()) + .build(); + backend.update( + table.nameIdentifier(), Entity.EntityType.TABLE, ignored -> movedTable); + return null; + } catch (Throwable throwable) { + return throwable; + } + }); + assertTrue(moveStarted.await(30, TimeUnit.SECONDS)); + // Resolving the target ID happens before the table transaction. The move must then wait on + // the target schema row instead of writing below a schema whose delete is about to commit. + assertThrows(TimeoutException.class, () -> moveResult.get(500, TimeUnit.MILLISECONDS)); + + allowDeleteCommit.countDown(); + Assertions.assertNull(deleteResult.get(30, TimeUnit.SECONDS)); + Assertions.assertInstanceOf( + NoSuchEntityException.class, moveResult.get(30, TimeUnit.SECONDS)); + } finally { + allowDeleteCommit.countDown(); + executor.shutdownNow(); + } + + TableEntity unchanged = + TableMetaService.getInstance().getTableByIdentifier(table.nameIdentifier()); + Assertions.assertEquals(table.namespace(), unchanged.namespace()); + assertFalse( + backend.exists( + NameIdentifier.of(metalakeName, catalogName, targetSchemaName, table.name()), + Entity.EntityType.TABLE)); + } + @TestTemplate public void testBatchGetTableByIdentifierIncludesVersionInfoFields() throws IOException { createAndInsertMakeLake(metalakeName); diff --git a/core/src/test/java/org/apache/gravitino/storage/relational/utils/TestPOConverters.java b/core/src/test/java/org/apache/gravitino/storage/relational/utils/TestPOConverters.java index 10e4ac042fe..994432dcaa5 100644 --- a/core/src/test/java/org/apache/gravitino/storage/relational/utils/TestPOConverters.java +++ b/core/src/test/java/org/apache/gravitino/storage/relational/utils/TestPOConverters.java @@ -700,6 +700,8 @@ public void testUpdateSchemaPOVersion() { assertEquals(1, initPO.getCurrentVersion()); assertEquals(1, initPO.getLastVersion()); assertEquals(0, initPO.getDeletedAt()); + assertEquals(2, updatePO.getCurrentVersion()); + assertEquals(2, updatePO.getLastVersion()); assertEquals("this is test2", updatePO.getSchemaComment()); }