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 f08c416d7db..ca85e04adab 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 @@ -39,6 +39,7 @@ import com.google.common.util.concurrent.ThreadFactoryBuilder; import java.io.FileNotFoundException; import java.io.IOException; +import java.io.UncheckedIOException; import java.time.Instant; import java.util.ArrayList; import java.util.Arrays; @@ -709,59 +710,75 @@ public Fileset alterFileset(NameIdentifier ident, FilesetChange... changes) @Override public boolean dropFileset(NameIdentifier ident) { try { - FilesetEntity filesetEntity = - store.get(ident, Entity.EntityType.FILESET, FilesetEntity.class); - - // For managed fileset, we should delete the related files. - if (!disableFSOps && filesetEntity.filesetType() == Fileset.Type.MANAGED) { - AtomicReference exception = new AtomicReference<>(); - Map storageLocations = - Maps.transformValues(filesetEntity.storageLocations(), Path::new); - storageLocations.forEach( - (locationName, location) -> { - try { - Map fsConf = - mergeUpLevelConfigurations(ident, filesetEntity.properties(), location); - FileSystem fs = getFileSystemWithCache(location, fsConf); - if (fs.exists(location)) { - if (!fs.delete(location, true)) { - LOG.warn( - "Failed to delete fileset {} location {} with location name {}", - ident, - location, - locationName); + // The relational store runs this cleanup after the metadata CAS wins but before committing + // its transaction. The callback therefore sees the exact deleted snapshot, and an I/O + // failure can still restore the metadata so the caller may fix permissions and retry. + // + // The price is that the recursive storage delete runs inside that transaction, holding the + // fileset rows and a pooled connection for as long as the filesystem takes. Dropping a + // fileset with a very large tree is therefore a slow write for that fileset, and enough + // concurrent drops can hold up the connection pool. + Optional deletedFileset = + store.deleteAndGet( + ident, + Entity.EntityType.FILESET, + FilesetEntity.class, + filesetEntity -> { + if (!disableFSOps && filesetEntity.filesetType() == Fileset.Type.MANAGED) { + try { + deleteManagedFilesetStorage(ident, filesetEntity); + } catch (IOException ioe) { + throw new UncheckedIOException(ioe); } - } else { - LOG.warn( - "Fileset {} location {} with location name {} does not exist", - ident, - location, - locationName); } - } catch (IOException ioe) { - LOG.warn( - "Failed to delete fileset {} location {} with location name {}", - ident, - location, - locationName, - ioe); - exception.set(ioe); - } - }); - if (exception.get() != null) { - throw exception.get(); - } - } - - return store.delete(ident, Entity.EntityType.FILESET); + }); + return deletedFileset.isPresent(); } catch (NoSuchEntityException ne) { LOG.warn("Fileset {} does not exist", ident); return false; + } catch (UncheckedIOException uioe) { + throw new RuntimeException("Failed to delete fileset " + ident, uioe.getCause()); } catch (IOException ioe) { throw new RuntimeException("Failed to delete fileset " + ident, ioe); } } + /** + * Removes the storage of a managed fileset while its metadata delete can still be rolled back. + * + *

The first location that cannot be removed stops the loop, so the drop is rejected before it + * takes away more data than it already has. The locations removed up to that point are gone for + * good, but attempting the remaining ones would only widen that gap. + */ + private void deleteManagedFilesetStorage(NameIdentifier ident, FilesetEntity filesetEntity) + throws IOException { + Map storageLocations = + Maps.transformValues(filesetEntity.storageLocations(), Path::new); + for (Map.Entry entry : storageLocations.entrySet()) { + String locationName = entry.getKey(); + Path location = entry.getValue(); + Map fsConf = + mergeUpLevelConfigurations(ident, filesetEntity.properties(), location); + FileSystem fs = getFileSystemWithCache(location, fsConf); + if (!fs.exists(location)) { + LOG.warn( + "Fileset {} location {} with location name {} does not exist", + ident, + location, + locationName); + continue; + } + if (!fs.delete(location, true) && fs.exists(location)) { + // A false return also covers a location that somebody else removed between the check above + // and this call. Only a location that is still there is a reason to reject the drop. + throw new IOException( + String.format( + "Failed to delete fileset %s location %s with location name %s", + ident, location, locationName)); + } + } + } + @Override public String getFileLocation(NameIdentifier ident, String subPath, String locationName) throws NoSuchFilesetException, NoSuchLocationNameException { 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 d4b70d3a8d9..1db30f01e21 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 @@ -76,6 +76,7 @@ import org.apache.gravitino.Catalog; import org.apache.gravitino.Config; import org.apache.gravitino.Configs; +import org.apache.gravitino.Entity; import org.apache.gravitino.EntityStore; import org.apache.gravitino.EntityStoreFactory; import org.apache.gravitino.GravitinoEnv; @@ -105,6 +106,7 @@ import org.apache.gravitino.exceptions.NoSuchFilesetException; import org.apache.gravitino.exceptions.NoSuchSchemaException; import org.apache.gravitino.exceptions.NonEmptySchemaException; +import org.apache.gravitino.exceptions.OptimisticLockException; import org.apache.gravitino.exceptions.SchemaAlreadyExistsException; import org.apache.gravitino.file.FileInfo; import org.apache.gravitino.file.Fileset; @@ -3386,6 +3388,115 @@ private static Stream testRenameArguments() { TEST_ROOT_PATH + "/fileset39")); } + @Test + public void testDropFilesetKeepsFilesWhenMetadataDropIsRejected() throws IOException { + String schemaName = "schema_drop_rejected"; + String filesetName = "fileset_drop_rejected"; + String catalogPath = TEST_ROOT_PATH + "/catalog_drop_rejected"; + createSchema(schemaName, "comment", catalogPath, null); + Fileset fileset = + createFileset(filesetName, schemaName, "comment", Fileset.Type.MANAGED, catalogPath, null); + + Path filesetPath = new Path(fileset.storageLocation()); + FileSystem fs = filesetPath.getFileSystem(new Configuration()); + Assertions.assertTrue(fs.exists(filesetPath)); + + NameIdentifier filesetIdent = NameIdentifier.of("m1", "c1", schemaName, filesetName); + EntityStore rejectingStore = Mockito.spy(store); + Mockito.doThrow(new OptimisticLockException("fileset was modified concurrently")) + .when(rejectingStore) + .deleteAndGet( + Mockito.eq(filesetIdent), + Mockito.eq(Entity.EntityType.FILESET), + Mockito.eq(FilesetEntity.class), + Mockito.any()); + + try (FilesetCatalogOperations ops = + new FilesetCatalogOperations(rejectingStore, secretManager)) { + ops.initialize( + ImmutableMap.of(LOCATION, catalogPath), + randomCatalogInfo("m1", "c1"), + FILESET_PROPERTIES_METADATA); + Assertions.assertThrows(OptimisticLockException.class, () -> ops.dropFileset(filesetIdent)); + } + + // The drop was refused, so the fileset row still advertises this location. Deleting the files + // anyway would leave that row pointing at data that is gone. + Assertions.assertTrue(fs.exists(filesetPath)); + Assertions.assertEquals( + filesetName, + store.get(filesetIdent, Entity.EntityType.FILESET, FilesetEntity.class).name()); + + fs.delete(filesetPath, true); + } + + @Test + public void testDropFilesetRollsBackMetadataWhenStorageDeletionFails() throws IOException { + String schemaName = "schema_drop_storage_failure"; + String filesetName = "fileset_drop_storage_failure"; + String catalogPath = TEST_ROOT_PATH + "/catalog_drop_storage_failure"; + createSchema(schemaName, "comment", catalogPath, null); + Fileset fileset = + createFileset(filesetName, schemaName, "comment", Fileset.Type.MANAGED, catalogPath, null); + NameIdentifier filesetIdent = NameIdentifier.of("m1", "c1", schemaName, filesetName); + + FileSystem failingFileSystem = Mockito.mock(FileSystem.class); + when(failingFileSystem.exists(any(Path.class))).thenReturn(true); + when(failingFileSystem.delete(any(Path.class), Mockito.eq(true))) + .thenThrow(new IOException("permission denied")); + + try (FilesetCatalogOperations ops = + Mockito.spy(new FilesetCatalogOperations(store, secretManager))) { + ops.initialize( + ImmutableMap.of(LOCATION, catalogPath), + randomCatalogInfo("m1", "c1"), + FILESET_PROPERTIES_METADATA); + doReturn(failingFileSystem).when(ops).getFileSystemWithCache(any(Path.class), any(Map.class)); + + RuntimeException failure = + Assertions.assertThrows(RuntimeException.class, () -> ops.dropFileset(filesetIdent)); + Assertions.assertTrue(failure.getMessage().contains("Failed to delete fileset")); + } + + // The failed physical cleanup aborts the outer transaction, so the fileset remains visible and + // a caller can repair its filesystem permissions and retry the drop. + FilesetEntity survivingFileset = + store.get(filesetIdent, Entity.EntityType.FILESET, FilesetEntity.class); + Assertions.assertEquals(fileset.storageLocation(), survivingFileset.storageLocation()); + + store.delete(filesetIdent, Entity.EntityType.FILESET); + new Path(fileset.storageLocation()) + .getFileSystem(new Configuration()) + .delete(new Path(fileset.storageLocation()), true); + } + + @Test + public void testDropFilesetSucceedsWhenTheLocationDisappearsFirst() throws IOException { + String schemaName = "schema_drop_vanished"; + String filesetName = "fileset_drop_vanished"; + String catalogPath = TEST_ROOT_PATH + "/catalog_drop_vanished"; + createSchema(schemaName, "comment", catalogPath, null); + Fileset fileset = + createFileset(filesetName, schemaName, "comment", Fileset.Type.MANAGED, catalogPath, null); + + Path filesetPath = new Path(fileset.storageLocation()); + FileSystem fs = filesetPath.getFileSystem(new Configuration()); + Assertions.assertTrue(fs.exists(filesetPath)); + // Somebody else removed the location already. A drop that finds nothing left to delete has + // nothing to complain about. + Assertions.assertTrue(fs.delete(filesetPath, true)); + + NameIdentifier filesetIdent = NameIdentifier.of("m1", "c1", schemaName, filesetName); + try (FilesetCatalogOperations ops = new FilesetCatalogOperations(store, secretManager)) { + ops.initialize( + ImmutableMap.of(LOCATION, catalogPath), + randomCatalogInfo("m1", "c1"), + FILESET_PROPERTIES_METADATA); + Assertions.assertTrue(ops.dropFileset(filesetIdent)); + Assertions.assertFalse(ops.dropFileset(filesetIdent), "fileset should be non-existent"); + } + } + private Schema createSchema(String name, String comment, String catalogPath, String schemaPath) throws IOException { return createSchema(name, comment, catalogPath, schemaPath, false); diff --git a/core/src/main/java/org/apache/gravitino/EntityStore.java b/core/src/main/java/org/apache/gravitino/EntityStore.java index 1f844a2e274..c63458c7208 100644 --- a/core/src/main/java/org/apache/gravitino/EntityStore.java +++ b/core/src/main/java/org/apache/gravitino/EntityStore.java @@ -23,6 +23,8 @@ import java.lang.reflect.Array; import java.util.Arrays; import java.util.List; +import java.util.Optional; +import java.util.function.Consumer; import java.util.function.Function; import org.apache.commons.lang3.tuple.Pair; import org.apache.gravitino.Entity.EntityType; @@ -221,6 +223,83 @@ default boolean delete(NameIdentifier ident, EntityType entityType) throws IOExc */ boolean delete(NameIdentifier ident, EntityType entityType, boolean cascade) throws IOException; + /** + * The only post-delete action an implementation that cannot run it before commit accepts. + * + *

Compared by reference, so a caller that supplies its own action reaches an implementation + * that honors the contract or gets told that this one cannot. + */ + Consumer NO_POST_DELETE_ACTION = ignored -> {}; + + /** + * Returns the shared no-op post-delete action. + * + * @param the entity type + * @return an action that does nothing + */ + @SuppressWarnings("unchecked") + static Consumer noPostDeleteAction() { + return (Consumer) NO_POST_DELETE_ACTION; + } + + /** + * Deletes an entity and returns the snapshot chosen by the delete operation. + * + *

The default implementation is intended for stores that serialize operations through {@link + * #executeInTransaction(Executable)}. Stores that can read and delete with one native + * compare-and-set should override this method so the returned snapshot is exactly the one that + * was deleted. + * + * @param ident the name identifier of the entity + * @param entityType the type of the entity + * @param clazz the concrete entity class + * @param the entity type + * @return the deleted entity, or empty when it did not exist + * @throws IOException if the delete operation fails + */ + default Optional deleteAndGet( + NameIdentifier ident, EntityType entityType, Class clazz) throws IOException { + return deleteAndGet(ident, entityType, clazz, noPostDeleteAction()); + } + + /** + * Deletes an entity, runs an action against the deleted snapshot, and returns that snapshot. + * + *

A transactional store should run the action after its delete has won but before committing. + * This lets callers couple non-database cleanup to the metadata transaction: an action failure + * can still roll the metadata delete back. + * + * @param ident the name identifier of the entity + * @param entityType the type of the entity + * @param clazz the concrete entity class + * @param postDeleteAction the action to run after deletion but before commit when supported + * @param the entity type + * @return the deleted entity, or empty when it did not exist + * @throws IOException if the delete operation fails + */ + default Optional deleteAndGet( + NameIdentifier ident, EntityType entityType, Class clazz, Consumer postDeleteAction) + throws IOException { + if (postDeleteAction != NO_POST_DELETE_ACTION) { + // This implementation can only run the action once the delete is committed, which is the + // opposite of what the contract promises. Refusing is better than silently leaving the + // caller with a committed delete and a failed cleanup. + throw new UnsupportedOperationException( + "This store cannot run a post-delete action while the delete can still be rolled back"); + } + + try { + E entity = get(ident, entityType, clazz); + if (!delete(ident, entityType)) { + return Optional.empty(); + } + postDeleteAction.accept(entity); + return Optional.of(entity); + } catch (NoSuchEntityException e) { + return Optional.empty(); + } + } + /** * Batch delete entities from the underlying storage by the specified list of {@link * org.apache.gravitino.NameIdentifier} and {@link EntityType}. diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/JDBCBackend.java b/core/src/main/java/org/apache/gravitino/storage/relational/JDBCBackend.java index 6661d0014d1..31ddb339723 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/JDBCBackend.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/JDBCBackend.java @@ -29,6 +29,8 @@ import java.util.Arrays; import java.util.List; import java.util.Map; +import java.util.Optional; +import java.util.function.Consumer; import java.util.function.Function; import java.util.stream.Collectors; import org.apache.commons.lang3.tuple.Pair; @@ -461,6 +463,49 @@ public boolean delete(NameIdentifier ident, Entity.EntityType entityType, boolea } } + @Override + public Optional deleteAndGet( + NameIdentifier ident, + Entity.EntityType entityType, + Class clazz, + Consumer postDeleteAction) + throws IOException { + if (entityType != Entity.EntityType.FILESET) { + return RelationalBackend.super.deleteAndGet(ident, entityType, clazz, postDeleteAction); + } + + boolean transactionOwner = !SessionUtils.isInTransaction(); + if (transactionOwner) { + SessionUtils.beginTransaction(); + } + boolean committed = false; + try { + FilesetEntity deletedFileset; + try { + deletedFileset = FilesetMetaService.getInstance().deleteFilesetAndGet(ident); + } catch (NoSuchEntityException e) { + // Only the delete itself may report the fileset as missing. A NoSuchEntityException from + // any later step means the delete did happen and something else failed, which must not be + // reported to the caller as "there was nothing to delete". + return Optional.empty(); + } + insertEntityChange(ident, entityType, OperateType.DROP); + E deletedEntity = clazz.cast(deletedFileset); + // Run external cleanup while the metadata delete can still be rolled back. The callback uses + // the same snapshot whose OCC token won above, so it cannot act on stale locations. + postDeleteAction.accept(deletedEntity); + if (transactionOwner) { + SessionUtils.commitTransaction(); + } + committed = true; + return Optional.of(deletedEntity); + } finally { + if (transactionOwner && !committed) { + SessionUtils.rollbackTransaction(); + } + } + } + @Override public int hardDeleteLegacyData(Entity.EntityType entityType, long legacyTimeline) throws IOException { diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/RelationalBackend.java b/core/src/main/java/org/apache/gravitino/storage/relational/RelationalBackend.java index 50e48af1c2f..ea5a8aa46c7 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/RelationalBackend.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/RelationalBackend.java @@ -21,11 +21,14 @@ import java.io.Closeable; import java.io.IOException; import java.util.List; +import java.util.Optional; +import java.util.function.Consumer; import java.util.function.Function; import org.apache.commons.lang3.tuple.Pair; import org.apache.gravitino.Config; import org.apache.gravitino.Entity; import org.apache.gravitino.EntityAlreadyExistsException; +import org.apache.gravitino.EntityStore; import org.apache.gravitino.HasIdentifier; import org.apache.gravitino.NameIdentifier; import org.apache.gravitino.Namespace; @@ -195,6 +198,61 @@ List batchGet( boolean delete(NameIdentifier ident, Entity.EntityType entityType, boolean cascade) throws IOException; + /** + * Deletes an entity and returns the snapshot used by that delete. + * + *

Backends with a native compare-and-set delete should override this method so the returned + * entity and the deleted row are based on the same read. + * + * @param ident the identifier of the entity + * @param entityType the entity type + * @param clazz the concrete entity class + * @param the concrete entity type + * @return the deleted entity, or empty when it did not exist + * @throws IOException if the store operation fails + */ + default Optional deleteAndGet( + NameIdentifier ident, Entity.EntityType entityType, Class clazz) throws IOException { + return deleteAndGet(ident, entityType, clazz, EntityStore.noPostDeleteAction()); + } + + /** + * Deletes an entity and runs an action against the deleted snapshot before commit when supported. + * + * @param ident the identifier of the entity + * @param entityType the entity type + * @param clazz the concrete entity class + * @param postDeleteAction the action to run after deletion but before commit when supported + * @param the concrete entity type + * @return the deleted entity, or empty when it did not exist + * @throws IOException if the store operation fails + */ + default Optional deleteAndGet( + NameIdentifier ident, + Entity.EntityType entityType, + Class clazz, + Consumer postDeleteAction) + throws IOException { + if (postDeleteAction != EntityStore.NO_POST_DELETE_ACTION) { + // This implementation can only run the action once the delete is committed, which is the + // opposite of what the contract promises. Refusing is better than silently leaving the + // caller with a committed delete and a failed cleanup. + throw new UnsupportedOperationException( + "This store cannot run a post-delete action while the delete can still be rolled back"); + } + + try { + E entity = get(ident, entityType); + if (!delete(ident, entityType, false)) { + return Optional.empty(); + } + postDeleteAction.accept(entity); + return Optional.of(entity); + } catch (NoSuchEntityException e) { + return Optional.empty(); + } + } + /** * Deletes the entities in the specified namespace and entity type. * diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/RelationalEntityStore.java b/core/src/main/java/org/apache/gravitino/storage/relational/RelationalEntityStore.java index bdac4f92b84..75e12ea8afa 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/RelationalEntityStore.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/RelationalEntityStore.java @@ -29,6 +29,7 @@ import java.util.List; import java.util.Optional; import java.util.concurrent.TimeUnit; +import java.util.function.Consumer; import java.util.function.Function; import javax.annotation.Nullable; import org.apache.commons.lang3.tuple.Pair; @@ -178,7 +179,14 @@ public boolean exists(NameIdentifier ident, Entity.EntityType entityType) throws public void put(E e, boolean overwritten) throws IOException, EntityAlreadyExistsException { backend.insert(e, overwritten); - cache.put(e); + if (overwritten) { + // An overwrite is resolved by the database, which may keep the identity and version of the + // row it already had. Caching the copy handed in here would publish values the stored row + // does not carry, so the next read is served from the backend instead. + cache.invalidate(e.nameIdentifier(), e.type()); + } else { + cache.put(e); + } } @Override @@ -320,6 +328,20 @@ public boolean delete(NameIdentifier ident, Entity.EntityType entityType, boolea } } + @Override + public Optional deleteAndGet( + NameIdentifier ident, + Entity.EntityType entityType, + Class clazz, + Consumer postDeleteAction) + throws IOException { + try { + return backend.deleteAndGet(ident, entityType, clazz, postDeleteAction); + } finally { + cache.invalidate(ident, entityType); + } + } + @Override public R executeInTransaction(Executable executable) { throw new UnsupportedOperationException("Unsupported operation in relational entity store."); diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/FilesetMetaMapper.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/FilesetMetaMapper.java index fcbbc66c0b8..fbd2ca9c266 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/FilesetMetaMapper.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/FilesetMetaMapper.java @@ -187,6 +187,19 @@ FilesetPO selectFilesetMetaBySchemaIdAndName( @SelectProvider(type = FilesetMetaSQLProviderFactory.class, method = "selectFilesetMetaById") FilesetPO selectFilesetMetaById(@Param("filesetId") Long filesetId); + /** + * Selects an active fileset metadata row by schema and name in the current transaction. + * + * @param schemaId the schema ID + * @param filesetName the fileset name + * @return the active fileset metadata, or {@code null} when it does not exist + */ + @SelectProvider( + type = FilesetMetaSQLProviderFactory.class, + method = "selectFilesetMetaBySchemaIdAndNameForUpdate") + FilesetPO selectFilesetMetaBySchemaIdAndNameForUpdate( + @Param("schemaId") Long schemaId, @Param("filesetName") String filesetName); + @Results({ @Result(property = "filesetId", column = "fileset_id", id = true), @Result(property = "filesetName", column = "fileset_name"), @@ -243,10 +256,18 @@ Integer updateFilesetMeta( method = "softDeleteFilesetMetasBySchemaIds") Integer softDeleteFilesetMetasBySchemaIds(@Param("schemaIds") List schemaIds); + /** + * Soft-deletes a fileset only if its version has not changed since the caller read it. + * + * @param filesetId the fileset ID + * @param currentVersion the version observed by the caller + * @return the number of deleted rows; zero means the fileset changed or disappeared + */ @UpdateProvider( type = FilesetMetaSQLProviderFactory.class, method = "softDeleteFilesetMetasByFilesetId") - Integer softDeleteFilesetMetasByFilesetId(@Param("filesetId") Long filesetId); + Integer softDeleteFilesetMetasByFilesetId( + @Param("filesetId") Long filesetId, @Param("currentVersion") Long currentVersion); @DeleteProvider( type = FilesetMetaSQLProviderFactory.class, diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/FilesetMetaSQLProviderFactory.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/FilesetMetaSQLProviderFactory.java index 07aaebb1cb2..b09dd069e80 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/FilesetMetaSQLProviderFactory.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/FilesetMetaSQLProviderFactory.java @@ -81,6 +81,18 @@ public static String selectFilesetMetaById(@Param("filesetId") Long filesetId) { return getProvider().selectFilesetMetaById(filesetId); } + /** + * Returns SQL that selects an active fileset metadata row by schema and name. + * + * @param schemaId the schema ID + * @param filesetName the fileset name + * @return the metadata-only select SQL + */ + public static String selectFilesetMetaBySchemaIdAndNameForUpdate( + @Param("schemaId") Long schemaId, @Param("filesetName") String filesetName) { + return getProvider().selectFilesetMetaBySchemaIdAndNameForUpdate(schemaId, filesetName); + } + public static String selectFilesetByFullQualifiedName( @Param("metalakeName") String metalakeName, @Param("catalogName") String catalogName, @@ -117,8 +129,16 @@ public static String softDeleteFilesetMetasBySchemaIds(@Param("schemaIds") List< return getProvider().softDeleteFilesetMetasBySchemaIds(schemaIds); } - public String softDeleteFilesetMetasByFilesetId(@Param("filesetId") Long filesetId) { - return getProvider().softDeleteFilesetMetasByFilesetId(filesetId); + /** + * Returns SQL that soft-deletes a fileset by ID and expected version. + * + * @param filesetId the fileset ID + * @param currentVersion the version observed by the caller + * @return the version-checked delete SQL + */ + public static String softDeleteFilesetMetasByFilesetId( + @Param("filesetId") Long filesetId, @Param("currentVersion") Long currentVersion) { + return getProvider().softDeleteFilesetMetasByFilesetId(filesetId, currentVersion); } public String deleteFilesetMetasByLegacyTimeline( diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/FilesetVersionMapper.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/FilesetVersionMapper.java index 931110faf06..2263b9f1408 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/FilesetVersionMapper.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/FilesetVersionMapper.java @@ -74,6 +74,15 @@ void insertFilesetVersionsOnDuplicateKeyUpdate( Integer deleteFilesetVersionsByLegacyTimeline( @Param("legacyTimeline") Long legacyTimeline, @Param("limit") int limit); + /** + * Returns the highest live version recorded for a fileset, or {@code null} when it has none. + * + * @param filesetId the fileset whose versions are inspected + * @return the highest version still present in the version table + */ + @SelectProvider(type = FilesetVersionSQLProviderFactory.class, method = "selectMaxFilesetVersion") + Long selectMaxFilesetVersion(@Param("filesetId") Long filesetId); + @SelectProvider( type = FilesetVersionSQLProviderFactory.class, method = "selectFilesetVersionsByRetentionCount") diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/FilesetVersionSQLProviderFactory.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/FilesetVersionSQLProviderFactory.java index 0d29d6f3f8f..eebb7e6802e 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/FilesetVersionSQLProviderFactory.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/FilesetVersionSQLProviderFactory.java @@ -83,6 +83,16 @@ public static String deleteFilesetVersionsByLegacyTimeline( return getProvider().deleteFilesetVersionsByLegacyTimeline(legacyTimeline, limit); } + /** + * Returns SQL that finds the highest active snapshot version for a fileset. + * + * @param filesetId the fileset ID + * @return the maximum-version query for the configured database + */ + public static String selectMaxFilesetVersion(@Param("filesetId") Long filesetId) { + return getProvider().selectMaxFilesetVersion(filesetId); + } + public static String selectFilesetVersionsByRetentionCount( @Param("versionRetentionCount") Long versionRetentionCount) { return getProvider().selectFilesetVersionsByRetentionCount(versionRetentionCount); diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/FilesetMetaBaseSQLProvider.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/FilesetMetaBaseSQLProvider.java index 8a1ad653c88..6a420fce1d2 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/FilesetMetaBaseSQLProvider.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/FilesetMetaBaseSQLProvider.java @@ -223,6 +223,29 @@ public String selectFilesetMetaById(@Param("filesetId") Long filesetId) { + " AND fm.deleted_at = 0 AND vi.deleted_at = 0"; } + /** + * Returns the active fileset metadata row selected by its natural key. + * + *

An overwrite may match the natural key instead of the incoming ID. Reading the stored row + * after the upsert tells dependent version rows which ID and database-generated version to use. + * + * @param schemaId the schema ID + * @param filesetName the fileset name + * @return the metadata-only select SQL + */ + public String selectFilesetMetaBySchemaIdAndNameForUpdate( + @Param("schemaId") Long schemaId, @Param("filesetName") String filesetName) { + return "SELECT fileset_id as filesetId, fileset_name as filesetName," + + " metalake_id as metalakeId, catalog_id as catalogId, schema_id as schemaId," + + " type as type, audit_info as auditInfo," + + " current_version as currentVersion, last_version as lastVersion," + + " deleted_at as deletedAt" + + " FROM " + + META_TABLE_NAME + + " WHERE schema_id = #{schemaId} AND fileset_name = #{filesetName}" + + " AND deleted_at = 0 FOR UPDATE"; + } + public String insertFilesetMeta(@Param("filesetMeta") FilesetPO filesetPO) { return "INSERT INTO " + META_TABLE_NAME @@ -268,11 +291,31 @@ public String insertFilesetMetaOnDuplicateKeyUpdate(@Param("filesetMeta") Filese + " schema_id = #{filesetMeta.schemaId}," + " type = #{filesetMeta.type}," + " audit_info = #{filesetMeta.auditInfo}," - + " current_version = #{filesetMeta.currentVersion}," - + " last_version = #{filesetMeta.lastVersion}," + // An overwrite is also a write observed by OCC. Advance from the stored value instead of + // resetting the row to the initial version carried by the incoming create request. + // + // Keep current_version last: MySQL evaluates these assignments left to right against the + // columns already assigned, while H2 and PostgreSQL evaluate every right-hand side against + // the row as it was before the update. Both agree only while current_version is read + // before it is assigned. + + " last_version = current_version + 1," + + " current_version = current_version + 1," + " deleted_at = #{filesetMeta.deletedAt}"; } + /** + * Returns SQL that updates a fileset only while its OCC version is unchanged and its next + * snapshot version is free. + * + *

The version is the concurrency token, so payload, name, and audit columns are deliberately + * excluded from the predicate. This also detects change-then-change-back races that a full-row + * comparison would miss. The snapshot check detects rows affected by the legacy overwrite bug + * without requiring a separate {@code MAX(version)} query on every normal alter. + * + * @param newFilesetPO the new fileset values + * @param oldFilesetPO the fileset values and version observed by the caller + * @return the version-checked update SQL + */ public String updateFilesetMeta( @Param("newFilesetMeta") FilesetPO newFilesetPO, @Param("oldFilesetMeta") FilesetPO oldFilesetPO) { @@ -288,15 +331,13 @@ public String updateFilesetMeta( + " last_version = #{newFilesetMeta.lastVersion}," + " deleted_at = #{newFilesetMeta.deletedAt}" + " WHERE fileset_id = #{oldFilesetMeta.filesetId}" - + " AND fileset_name = #{oldFilesetMeta.filesetName}" - + " AND metalake_id = #{oldFilesetMeta.metalakeId}" - + " AND catalog_id = #{oldFilesetMeta.catalogId}" - + " AND schema_id = #{oldFilesetMeta.schemaId}" - + " AND type = #{oldFilesetMeta.type}" - + " AND audit_info = #{oldFilesetMeta.auditInfo}" + " AND current_version = #{oldFilesetMeta.currentVersion}" - + " AND last_version = #{oldFilesetMeta.lastVersion}" - + " AND deleted_at = 0"; + + " AND deleted_at = 0" + + " AND NOT EXISTS (SELECT 1 FROM " + + VERSION_TABLE_NAME + + " fv WHERE fv.fileset_id = #{oldFilesetMeta.filesetId}" + + " AND fv.version >= #{newFilesetMeta.currentVersion}" + + " AND fv.deleted_at = 0)"; } public String softDeleteFilesetMetasByMetalakeId(@Param("metalakeId") Long metalakeId) { @@ -329,12 +370,21 @@ public String softDeleteFilesetMetasBySchemaIds(@Param("schemaIds") List s + ""; } - public String softDeleteFilesetMetasByFilesetId(@Param("filesetId") Long filesetId) { + /** + * Returns SQL that deletes only the fileset version observed by the caller. + * + * @param filesetId the fileset ID + * @param currentVersion the version observed by the caller + * @return the version-checked delete SQL + */ + public String softDeleteFilesetMetasByFilesetId( + @Param("filesetId") Long filesetId, @Param("currentVersion") Long currentVersion) { return "UPDATE " + META_TABLE_NAME + " SET deleted_at = (UNIX_TIMESTAMP() * 1000.0)" + " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000" - + " WHERE fileset_id = #{filesetId} AND deleted_at = 0"; + + " WHERE fileset_id = #{filesetId}" + + " AND current_version = #{currentVersion} AND deleted_at = 0"; } public String deleteFilesetMetasByLegacyTimeline( diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/FilesetVersionBaseSQLProvider.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/FilesetVersionBaseSQLProvider.java index 12204fe0cef..61e013ac1cc 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/FilesetVersionBaseSQLProvider.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/FilesetVersionBaseSQLProvider.java @@ -116,6 +116,19 @@ public String deleteFilesetVersionsByLegacyTimeline( + " WHERE deleted_at > 0 AND deleted_at < #{legacyTimeline} LIMIT #{limit}"; } + /** + * Returns SQL that finds the highest active snapshot version owned by a fileset. + * + * @param filesetId the fileset ID + * @return the maximum-version query + */ + public String selectMaxFilesetVersion(@Param("filesetId") Long filesetId) { + return "SELECT MAX(version)" + + " FROM " + + VERSION_TABLE_NAME + + " WHERE fileset_id = #{filesetId} AND deleted_at = 0"; + } + public String selectFilesetVersionsByRetentionCount( @Param("versionRetentionCount") Long versionRetentionCount) { return "SELECT fileset_id as filesetId," diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/FilesetMetaPostgreSQLProvider.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/FilesetMetaPostgreSQLProvider.java index ebb4d1731fe..84e53bb79ae 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/FilesetMetaPostgreSQLProvider.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/FilesetMetaPostgreSQLProvider.java @@ -57,11 +57,12 @@ public String softDeleteFilesetMetasBySchemaIds(List schemaIds) { } @Override - public String softDeleteFilesetMetasByFilesetId(Long filesetId) { + public String softDeleteFilesetMetasByFilesetId(Long filesetId, Long currentVersion) { return "UPDATE " + META_TABLE_NAME + " SET deleted_at = CAST(EXTRACT(EPOCH FROM CURRENT_TIMESTAMP) * 1000 AS BIGINT)" - + " WHERE fileset_id = #{filesetId} AND deleted_at = 0"; + + " WHERE fileset_id = #{filesetId}" + + " AND current_version = #{currentVersion} AND deleted_at = 0"; } @Override @@ -93,15 +94,23 @@ public String insertFilesetMetaOnDuplicateKeyUpdate(FilesetPO filesetPO) { + " #{filesetMeta.lastVersion}," + " #{filesetMeta.deletedAt}" + " )" - + " ON CONFLICT(fileset_id) DO UPDATE SET" + // Overwrite is selected by name, and a create request normally carries a newly generated + // ID. Target the natural key so PostgreSQL preserves the ID of the row being replaced, the + // same behavior that MySQL and H2 provide for their duplicate-key upsert. + + " ON CONFLICT(schema_id, fileset_name, deleted_at) DO UPDATE SET" + " fileset_name = #{filesetMeta.filesetName}," + " metalake_id = #{filesetMeta.metalakeId}," + " catalog_id = #{filesetMeta.catalogId}," + " schema_id = #{filesetMeta.schemaId}," + " type = #{filesetMeta.type}," + " audit_info = #{filesetMeta.auditInfo}," - + " current_version = #{filesetMeta.currentVersion}," - + " last_version = #{filesetMeta.lastVersion}," + // PostgreSQL requires the stored row to be qualified on the update side of ON CONFLICT. + + " current_version = " + + META_TABLE_NAME + + ".current_version + 1," + + " last_version = " + + META_TABLE_NAME + + ".current_version + 1," + " deleted_at = #{filesetMeta.deletedAt}"; } } 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 b29d1981931..70ce93113aa 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 @@ -22,9 +22,12 @@ import com.google.common.base.Preconditions; import java.io.IOException; +import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.Objects; -import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; import java.util.function.Function; import java.util.stream.Collectors; import org.apache.gravitino.Entity; @@ -33,6 +36,7 @@ import org.apache.gravitino.MetadataObject; import org.apache.gravitino.NameIdentifier; import org.apache.gravitino.Namespace; +import org.apache.gravitino.StringIdentifier; import org.apache.gravitino.exceptions.NoSuchEntityException; import org.apache.gravitino.meta.FilesetEntity; import org.apache.gravitino.meta.NamespacedEntityId; @@ -163,8 +167,10 @@ public void insertFileset(FilesetEntity filesetEntity, boolean overwrite) throws fillFilesetPOBuilderParentEntityId(builder, filesetEntity.namespace()); FilesetPO po = POConverters.initializeFilesetPOWithVersion(filesetEntity, builder); + AtomicReference persistedPO = new AtomicReference<>(po); - // insert both fileset meta table and version table + // The schema lock, metadata row, and every storage-location version row share one + // transaction. A failure in any later step restores all earlier writes. SessionUtils.doMultipleWithCommit( // Hold the parent schema row until this transaction ends, so the fileset cannot be // written below a schema that is being dropped. @@ -179,22 +185,42 @@ public void insertFileset(FilesetEntity filesetEntity, boolean overwrite) throws SessionUtils.doWithoutCommit( FilesetMetaMapper.class, mapper -> { - if (overwrite) { - mapper.insertFilesetMetaOnDuplicateKeyUpdate(po); - } else { + FilesetPO storedPO = + overwrite + ? mapper.selectFilesetMetaBySchemaIdAndNameForUpdate( + po.getSchemaId(), po.getFilesetName()) + : null; + if (storedPO == null) { mapper.insertFilesetMeta(po); + return; } + + // Resolve the natural-key overwrite before building its snapshot. This keeps + // the stored ID in both the metadata row and identifier property without a + // post-insert JSON/PO rewrite. + FilesetEntity replacement = + filesetWithPersistedId(filesetEntity, storedPO.getFilesetId()); + Long maxStoredVersion = + SessionUtils.getWithoutCommit( + FilesetVersionMapper.class, + versionMapper -> + versionMapper.selectMaxFilesetVersion(storedPO.getFilesetId())); + FilesetPO replacementPO = + POConverters.updateFilesetPOWithVersion( + storedPO, replacement, maxStoredVersion); + Integer updated = mapper.updateFilesetMeta(replacementPO, storedPO); + Preconditions.checkState( + updated != null && updated == 1, + "The overwritten fileset %s in schema %s changed while its row was held", + po.getFilesetName(), + po.getSchemaId()); + persistedPO.set(replacementPO); }), () -> SessionUtils.doWithoutCommit( FilesetVersionMapper.class, - mapper -> { - if (overwrite) { - mapper.insertFilesetVersionsOnDuplicateKeyUpdate(po.getFilesetVersionPOs()); - } else { - mapper.insertFilesetVersions(po.getFilesetVersionPOs()); - } - })); + mapper -> + mapper.insertFilesetVersions(persistedPO.get().getFilesetVersionPOs()))); } catch (RuntimeException re) { ExceptionUtils.checkSQLException( re, Entity.EntityType.FILESET, filesetEntity.nameIdentifier().toString()); @@ -213,120 +239,68 @@ public FilesetEntity updateFileset( FilesetEntity newEntity = (FilesetEntity) updater.apply((E) oldFilesetEntity); Preconditions.checkArgument( Objects.equals(oldFilesetEntity.id(), newEntity.id()), - "The updated fileset entity id: %s should be same with the table entity id before: %s", + "The updated fileset entity id: %s should be same with the fileset entity id before: %s", newEntity.id(), oldFilesetEntity.id()); - Integer updateResult; try { - boolean checkNeedUpdateVersion = - POConverters.checkFilesetVersionNeedUpdate( - oldFilesetPO.getFilesetVersionPOs(), newEntity); FilesetPO newFilesetPO = - POConverters.updateFilesetPOWithVersion(oldFilesetPO, newEntity, checkNeedUpdateVersion); - if (checkNeedUpdateVersion) { - // These operations are performed atomically within a single transaction. The version - // insert is protected by a unique constraint on `fileset_id + version + deleted_at`. If - // the meta update affects 0 rows (concurrent modification), the transaction is rolled - // back — including the version insert — and the update is treated as a conflict. - int[] metaUpdateCountRef = new int[1]; - try { - SessionUtils.doMultipleWithCommit( - () -> - SessionUtils.doWithoutCommit( - FilesetVersionMapper.class, - mapper -> mapper.insertFilesetVersions(newFilesetPO.getFilesetVersionPOs())), - () -> { - metaUpdateCountRef[0] = - SessionUtils.getWithoutCommit( - FilesetMetaMapper.class, - mapper -> mapper.updateFilesetMeta(newFilesetPO, oldFilesetPO)); - if (metaUpdateCountRef[0] == 0) { - throw new RuntimeException("Failed to update the entity: " + identifier); - } - }); - updateResult = 1; - } catch (RuntimeException re) { - if (metaUpdateCountRef[0] == 0) { - // The meta update matched no rows; the transaction was rolled back, - // including the version insert above. - throw new IOException("Failed to update the entity: " + identifier); - } else { - ExceptionUtils.checkSQLException( - re, Entity.EntityType.FILESET, newEntity.nameIdentifier().toString()); - throw re; - } - } - } else { - int[] metaUpdateCountRef = new int[1]; - SessionUtils.doMultipleWithCommit( - () -> - metaUpdateCountRef[0] = - SessionUtils.getWithoutCommit( - FilesetMetaMapper.class, - mapper -> mapper.updateFilesetMeta(newFilesetPO, oldFilesetPO))); - updateResult = metaUpdateCountRef[0]; + POConverters.updateFilesetPOWithVersion(oldFilesetPO, newEntity, null); + if (tryUpdateFileset(newFilesetPO, oldFilesetPO)) { + return newEntity; + } + + // The metadata CAS also rejects a version that already has an active stored snapshot. Only + // that uncommon legacy case needs the MAX(version) round trip; normal alters finish above. + Long maxStoredVersion = + SessionUtils.getWithoutCommit( + FilesetVersionMapper.class, + mapper -> mapper.selectMaxFilesetVersion(oldFilesetPO.getFilesetId())); + if (maxStoredVersion != null + && maxStoredVersion >= newFilesetPO.getCurrentVersion() + && tryUpdateFileset( + POConverters.updateFilesetPOWithVersion(oldFilesetPO, newEntity, maxStoredVersion), + oldFilesetPO)) { + return newEntity; } + + throw filesetWriteFailure(identifier, oldFilesetPO); } catch (RuntimeException re) { ExceptionUtils.checkSQLException( re, Entity.EntityType.FILESET, newEntity.nameIdentifier().toString()); throw re; } - - if (updateResult > 0) { - return newEntity; - } else { - throw new IOException("Failed to update the entity: " + identifier); - } } @Monitored( metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME, baseMetricName = "deleteFileset") public boolean deleteFileset(NameIdentifier identifier) { + deleteFilesetAndGet(identifier); + return true; + } + + /** + * Deletes a fileset and returns the exact entity snapshot protected by the delete CAS. + * + *

Callers that also remove managed storage use this snapshot rather than a separate earlier + * read. Otherwise an alter between the two reads could make metadata deletion succeed for new + * locations while physical cleanup still deletes the old locations. + * + * @param identifier the fileset identifier + * @return the fileset snapshot that was deleted + */ + public FilesetEntity deleteFilesetAndGet(NameIdentifier identifier) { FilesetPO filesetPO = getFilesetPOByIdentifier(identifier); - Long filesetId = filesetPO.getFilesetId(); + FilesetEntity deletedFileset = POConverters.fromFilesetPO(filesetPO, identifier.namespace()); - // We should delete meta and version info - AtomicInteger deleteResult = new AtomicInteger(0); + // Delete the root row first and only if it still has the version we read. A stale drop stops + // before it can remove versions, tags, policies, or any other related data. SessionUtils.doMultipleWithCommit( - () -> - deleteResult.set( - SessionUtils.getWithoutCommit( - FilesetMetaMapper.class, - mapper -> mapper.softDeleteFilesetMetasByFilesetId(filesetId))), - () -> { - if (deleteResult.get() > 0) { - SessionUtils.doWithoutCommit( - FilesetVersionMapper.class, - mapper -> mapper.softDeleteFilesetVersionsByFilesetId(filesetId)); - SessionUtils.doWithoutCommit( - OwnerMetaMapper.class, - mapper -> - mapper.softDeleteOwnerRelByMetadataObjectIdAndType( - filesetId, MetadataObject.Type.FILESET.name())); - SessionUtils.doWithoutCommit( - SecurableObjectMapper.class, - mapper -> - mapper.softDeleteObjectRelsByMetadataObject( - filesetId, MetadataObject.Type.FILESET.name())); - SessionUtils.doWithoutCommit( - TagMetadataObjectRelMapper.class, - mapper -> - mapper.softDeleteTagMetadataObjectRelsByMetadataObject( - filesetId, MetadataObject.Type.FILESET.name())); - SessionUtils.doWithoutCommit( - StatisticMetaMapper.class, - mapper -> mapper.softDeleteStatisticsByEntityId(filesetId)); - SessionUtils.doWithoutCommit( - PolicyMetadataObjectRelMapper.class, - mapper -> - mapper.softDeletePolicyMetadataObjectRelsByMetadataObject( - filesetId, MetadataObject.Type.FILESET.name())); - } - }); + () -> deleteFilesetWithVersion(identifier, filesetPO), + () -> deleteFilesetDependents(filesetPO.getFilesetId())); - return deleteResult.get() > 0; + return deletedFileset; } @Monitored( @@ -480,4 +454,117 @@ public List batchGetFilesetByIdentifier(List iden return POConverters.fromFilesetPOs(filesetPOs, firstIdent.namespace()); }); } + + /** + * Soft-deletes the observed fileset metadata row without starting a transaction. + * + *

The caller must run this method in the same transaction as dependent cleanup. Package access + * also lets concurrency tests submit a deliberately stale snapshot without duplicating the + * production CAS logic. + * + * @param identifier the fileset identity observed by the caller + * @param observedFilesetPO the fileset row and OCC version observed by the caller + */ + void deleteFilesetWithVersion(NameIdentifier identifier, FilesetPO observedFilesetPO) { + int deleted = + SessionUtils.getWithoutCommit( + FilesetMetaMapper.class, + mapper -> + mapper.softDeleteFilesetMetasByFilesetId( + observedFilesetPO.getFilesetId(), observedFilesetPO.getCurrentVersion())); + if (deleted == 0) { + throw filesetWriteFailure(identifier, observedFilesetPO); + } + } + + private boolean tryUpdateFileset(FilesetPO newFilesetPO, FilesetPO oldFilesetPO) { + AtomicBoolean updated = new AtomicBoolean(false); + SessionUtils.doMultipleWithCommit( + () -> { + Integer updateCount = + SessionUtils.getWithoutCommit( + FilesetMetaMapper.class, + mapper -> mapper.updateFilesetMeta(newFilesetPO, oldFilesetPO)); + updated.set(updateCount != null && updateCount > 0); + }, + () -> { + if (updated.get()) { + // The metadata row now points to this complete snapshot. It stays in the same + // transaction so a failed version insert also restores the metadata version. + SessionUtils.doWithoutCommit( + FilesetVersionMapper.class, + mapper -> mapper.insertFilesetVersions(newFilesetPO.getFilesetVersionPOs())); + } + }); + return updated.get(); + } + + private FilesetEntity filesetWithPersistedId(FilesetEntity filesetEntity, Long persistedId) { + Map properties = filesetEntity.properties(); + if (properties != null && properties.containsKey(StringIdentifier.ID_KEY)) { + properties = new HashMap<>(properties); + properties.put(StringIdentifier.ID_KEY, StringIdentifier.fromId(persistedId).toString()); + } + + return FilesetEntity.builder() + .withId(persistedId) + .withName(filesetEntity.name()) + .withNamespace(filesetEntity.namespace()) + .withComment(filesetEntity.comment()) + .withFilesetType(filesetEntity.filesetType()) + .withStorageLocations(filesetEntity.storageLocations()) + .withProperties(properties) + .withAuditInfo(filesetEntity.auditInfo()) + .build(); + } + + private void deleteFilesetDependents(Long filesetId) { + // The fileset row has already passed its version check. All cleanup below uses the same + // transaction, so either the root and every related row are deleted together, or none are. + SessionUtils.doWithoutCommit( + FilesetVersionMapper.class, + mapper -> mapper.softDeleteFilesetVersionsByFilesetId(filesetId)); + SessionUtils.doWithoutCommit( + OwnerMetaMapper.class, + mapper -> + mapper.softDeleteOwnerRelByMetadataObjectIdAndType( + filesetId, MetadataObject.Type.FILESET.name())); + SessionUtils.doWithoutCommit( + SecurableObjectMapper.class, + mapper -> + mapper.softDeleteObjectRelsByMetadataObject( + filesetId, MetadataObject.Type.FILESET.name())); + SessionUtils.doWithoutCommit( + TagMetadataObjectRelMapper.class, + mapper -> + mapper.softDeleteTagMetadataObjectRelsByMetadataObject( + filesetId, MetadataObject.Type.FILESET.name())); + SessionUtils.doWithoutCommit( + StatisticMetaMapper.class, mapper -> mapper.softDeleteStatisticsByEntityId(filesetId)); + SessionUtils.doWithoutCommit( + PolicyMetadataObjectRelMapper.class, + mapper -> + mapper.softDeletePolicyMetadataObjectRelsByMetadataObject( + filesetId, MetadataObject.Type.FILESET.name())); + } + + private RuntimeException filesetWriteFailure( + NameIdentifier identifier, FilesetPO observedFilesetPO) { + // The failed CAS has already serialized with an in-flight writer. A non-locking natural-key + // lookup is enough to distinguish a disappeared name from one that still names either the + // modified fileset or a replacement, without holding another row lock on the failure path. + Long currentFilesetId = + SessionUtils.getWithoutCommit( + FilesetMetaMapper.class, + mapper -> + mapper.selectFilesetIdBySchemaIdAndName( + observedFilesetPO.getSchemaId(), observedFilesetPO.getFilesetName())); + if (currentFilesetId == null) { + return new NoSuchEntityException( + NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE, + Entity.EntityType.FILESET.name().toLowerCase(), + identifier.name()); + } + return ExceptionUtils.concurrentModification(Entity.EntityType.FILESET, identifier); + } } 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 e166a9f4c6c..5557378eda4 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 @@ -29,6 +29,7 @@ import java.util.Map; import java.util.Objects; import java.util.stream.Collectors; +import javax.annotation.Nullable; import org.apache.commons.lang3.StringUtils; import org.apache.gravitino.Catalog; import org.apache.gravitino.MetadataObject; @@ -701,43 +702,47 @@ public static FilesetPO initializeFilesetPOWithVersion( * * @param oldFilesetPO the existing {@link FilesetPO} containing the current and last version data * @param newFileset the {@link FilesetEntity} with updated metadata and storage locations - * @param needUpdateVersion true to increment and update version fields; false to keep versions - * unchanged + * @param maxStoredVersion the highest version the fileset still has a stored snapshot for, or + * {@code null} when it has none * @return {@code FilesetPO} object with updated version * @throws RuntimeException if JSON serialization of properties fails */ public static FilesetPO updateFilesetPOWithVersion( - FilesetPO oldFilesetPO, FilesetEntity newFileset, boolean needUpdateVersion) { + FilesetPO oldFilesetPO, FilesetEntity newFileset, @Nullable Long maxStoredVersion) { try { - Long lastVersion = oldFilesetPO.getLastVersion(); - Long currentVersion; - List newFilesetVersionPOs; - // Will set the version to the last version + 1 - if (needUpdateVersion) { - lastVersion++; - currentVersion = lastVersion; - String props = JsonUtils.anyFieldMapper().writeValueAsString(newFileset.properties()); - newFilesetVersionPOs = - newFileset.storageLocations().entrySet().stream() - .map( - entry -> - FilesetVersionPO.builder() - .withMetalakeId(oldFilesetPO.getMetalakeId()) - .withCatalogId(oldFilesetPO.getCatalogId()) - .withSchemaId(oldFilesetPO.getSchemaId()) - .withFilesetId(newFileset.id()) - .withVersion(currentVersion) - .withFilesetComment(newFileset.comment()) - .withLocationName(entry.getKey()) - .withStorageLocation(entry.getValue()) - .withProperties(props) - .withDeletedAt(DEFAULT_DELETED_AT) - .build()) - .collect(Collectors.toList()); - } else { - currentVersion = oldFilesetPO.getCurrentVersion(); - newFilesetVersionPOs = oldFilesetPO.getFilesetVersionPOs(); + // Every successful fileset alter advances the OCC token. The current version is also the + // value used by reads to find the fileset details, so even a rename or audit-only change + // needs a complete snapshot at the new version. Alters that change nothing therefore still + // write one row per storage location; the version retention job is what removes them again. + // + // The stored snapshots are taken into account as well, because a fileset written before the + // version reset was fixed can carry snapshots newer than the version its metadata row + // records. Starting from the metadata row alone would rebuild a version that already exists + // and collide with the unique key over (fileset_id, version, storage_location_name). + long previousVersion = + Math.max(oldFilesetPO.getLastVersion(), oldFilesetPO.getCurrentVersion()); + if (maxStoredVersion != null) { + previousVersion = Math.max(previousVersion, maxStoredVersion); } + Long currentVersion = previousVersion + 1; + String props = JsonUtils.anyFieldMapper().writeValueAsString(newFileset.properties()); + List newFilesetVersionPOs = + newFileset.storageLocations().entrySet().stream() + .map( + entry -> + FilesetVersionPO.builder() + .withMetalakeId(oldFilesetPO.getMetalakeId()) + .withCatalogId(oldFilesetPO.getCatalogId()) + .withSchemaId(oldFilesetPO.getSchemaId()) + .withFilesetId(newFileset.id()) + .withVersion(currentVersion) + .withFilesetComment(newFileset.comment()) + .withLocationName(entry.getKey()) + .withStorageLocation(entry.getValue()) + .withProperties(props) + .withDeletedAt(DEFAULT_DELETED_AT) + .build()) + .collect(Collectors.toList()); return FilesetPO.builder() .withFilesetId(newFileset.id()) .withFilesetName(newFileset.name()) @@ -747,7 +752,7 @@ public static FilesetPO updateFilesetPOWithVersion( .withType(newFileset.filesetType().name()) .withAuditInfo(JsonUtils.anyFieldMapper().writeValueAsString(newFileset.auditInfo())) .withCurrentVersion(currentVersion) - .withLastVersion(lastVersion) + .withLastVersion(currentVersion) .withDeletedAt(DEFAULT_DELETED_AT) .withFilesetVersionPOs(newFilesetVersionPOs) .build(); @@ -756,31 +761,6 @@ public static FilesetPO updateFilesetPOWithVersion( } } - public static boolean checkFilesetVersionNeedUpdate( - List oldFilesetVersionPOs, FilesetEntity newFileset) { - Map storageLocations = - oldFilesetVersionPOs.stream() - .collect( - Collectors.toMap( - FilesetVersionPO::getLocationName, FilesetVersionPO::getStorageLocation)); - if (!StringUtils.equals(oldFilesetVersionPOs.get(0).getFilesetComment(), newFileset.comment()) - || !Objects.equals(storageLocations, newFileset.storageLocations())) { - return true; - } - - try { - Map oldProperties = - JsonUtils.anyFieldMapper() - .readValue(oldFilesetVersionPOs.get(0).getProperties(), Map.class); - if (oldProperties == null) { - return newFileset.properties() != null; - } - return !oldProperties.equals(newFileset.properties()); - } catch (JsonProcessingException e) { - throw new RuntimeException("Failed to deserialize json object:", e); - } - } - public static boolean checkPolicyVersionNeedUpdate( PolicyVersionPO oldPolicyVersionPO, PolicyEntity newPolicy) { if (!StringUtils.equals(oldPolicyVersionPO.getPolicyComment(), newPolicy.comment()) diff --git a/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/base/TestFilesetMetaBaseSQLProvider.java b/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/base/TestFilesetMetaBaseSQLProvider.java new file mode 100644 index 00000000000..cdbfcffd379 --- /dev/null +++ b/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/base/TestFilesetMetaBaseSQLProvider.java @@ -0,0 +1,77 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.gravitino.storage.relational.mapper.provider.base; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class TestFilesetMetaBaseSQLProvider { + + private static final FilesetMetaBaseSQLProvider PROVIDER = new FilesetMetaBaseSQLProvider(); + + @Test + void testOverwriteAdvancesStoredVersion() { + String sql = PROVIDER.insertFilesetMetaOnDuplicateKeyUpdate(null); + String updateClause = sql.substring(sql.indexOf(" ON DUPLICATE KEY UPDATE")); + + Assertions.assertTrue(updateClause.contains("last_version = current_version + 1")); + Assertions.assertTrue(updateClause.contains("current_version = current_version + 1")); + Assertions.assertTrue( + updateClause.indexOf("last_version =") < updateClause.indexOf("current_version =")); + Assertions.assertFalse( + updateClause.contains("current_version = #{filesetMeta.currentVersion}")); + Assertions.assertFalse(updateClause.contains("last_version = #{filesetMeta.lastVersion}")); + } + + @Test + void testUpdateUsesVersionCasAndRejectsAnOccupiedSnapshotVersion() { + String sql = PROVIDER.updateFilesetMeta(null, null); + String whereClause = sql.substring(sql.indexOf(" WHERE")); + + Assertions.assertEquals( + " WHERE fileset_id = #{oldFilesetMeta.filesetId}" + + " AND current_version = #{oldFilesetMeta.currentVersion}" + + " AND deleted_at = 0" + + " AND NOT EXISTS (SELECT 1 FROM fileset_version_info fv" + + " WHERE fv.fileset_id = #{oldFilesetMeta.filesetId}" + + " AND fv.version >= #{newFilesetMeta.currentVersion}" + + " AND fv.deleted_at = 0)", + whereClause); + } + + @Test + void testDirectDeleteUsesVersionCas() { + String sql = PROVIDER.softDeleteFilesetMetasByFilesetId(null, null); + + Assertions.assertTrue(sql.contains("AND current_version = #{currentVersion}")); + Assertions.assertTrue(sql.endsWith("AND deleted_at = 0")); + } + + @Test + void testOverwriteReadUsesNaturalKeyAndMetadataOnly() { + String sql = PROVIDER.selectFilesetMetaBySchemaIdAndNameForUpdate(null, null); + + Assertions.assertTrue( + sql.contains( + "WHERE schema_id = #{schemaId} AND fileset_name = #{filesetName}" + + " AND deleted_at = 0")); + Assertions.assertFalse(sql.contains("fileset_version_info")); + Assertions.assertTrue(sql.endsWith("FOR UPDATE")); + } +} diff --git a/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/TestFilesetMetaPostgreSQLProvider.java b/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/TestFilesetMetaPostgreSQLProvider.java new file mode 100644 index 00000000000..49753bfdea2 --- /dev/null +++ b/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/TestFilesetMetaPostgreSQLProvider.java @@ -0,0 +1,51 @@ +/* + * 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 org.apache.gravitino.storage.relational.mapper.FilesetMetaMapper; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class TestFilesetMetaPostgreSQLProvider { + + @Test + void testOverwriteAdvancesStoredVersion() { + String sql = new FilesetMetaPostgreSQLProvider().insertFilesetMetaOnDuplicateKeyUpdate(null); + String conflictClause = sql.substring(sql.indexOf(" ON CONFLICT")); + + Assertions.assertTrue( + conflictClause.startsWith(" ON CONFLICT(schema_id, fileset_name, deleted_at)")); + Assertions.assertTrue( + conflictClause.contains( + "current_version = " + FilesetMetaMapper.META_TABLE_NAME + ".current_version + 1")); + Assertions.assertTrue( + conflictClause.contains( + "last_version = " + FilesetMetaMapper.META_TABLE_NAME + ".current_version + 1")); + Assertions.assertFalse(conflictClause.contains("#{filesetMeta.currentVersion}")); + Assertions.assertFalse(conflictClause.contains("#{filesetMeta.lastVersion}")); + } + + @Test + void testDirectDeleteUsesVersionCas() { + String sql = new FilesetMetaPostgreSQLProvider().softDeleteFilesetMetasByFilesetId(null, null); + + Assertions.assertTrue(sql.contains("AND current_version = #{currentVersion}")); + Assertions.assertTrue(sql.endsWith("AND deleted_at = 0")); + } +} diff --git a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestFilesetMetaService.java b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestFilesetMetaService.java index 677c6ada5a2..52550651941 100644 --- a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestFilesetMetaService.java +++ b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestFilesetMetaService.java @@ -36,7 +36,9 @@ import java.util.List; import java.util.Map; import java.util.Set; +import java.util.function.Function; import java.util.stream.Collectors; +import org.apache.commons.lang3.StringUtils; import org.apache.commons.lang3.reflect.FieldUtils; import org.apache.commons.lang3.tuple.Pair; import org.apache.gravitino.Config; @@ -46,14 +48,22 @@ import org.apache.gravitino.GravitinoEnv; import org.apache.gravitino.NameIdentifier; import org.apache.gravitino.Namespace; +import org.apache.gravitino.StringIdentifier; import org.apache.gravitino.exceptions.NoSuchEntityException; +import org.apache.gravitino.exceptions.OptimisticLockException; import org.apache.gravitino.file.Fileset; import org.apache.gravitino.integration.test.util.GravitinoITUtils; import org.apache.gravitino.meta.AuditInfo; import org.apache.gravitino.meta.FilesetEntity; +import org.apache.gravitino.meta.SchemaEntity; import org.apache.gravitino.storage.RandomIdGenerator; import org.apache.gravitino.storage.relational.TestJDBCBackend; +import org.apache.gravitino.storage.relational.mapper.FilesetMetaMapper; +import org.apache.gravitino.storage.relational.mapper.FilesetVersionMapper; +import org.apache.gravitino.storage.relational.po.FilesetPO; +import org.apache.gravitino.storage.relational.po.FilesetVersionPO; import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper; +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; @@ -392,8 +402,7 @@ private FilesetEntity createFilesetEntity( } @TestTemplate - public void testUpdateFilesetReturnsSuccessWhenVersionedMetaUpdateAffectsNoRows() - throws IOException { + public void testAlterReportsOptimisticLockConflictAndKeepsWinnerVersion() throws IOException { String filesetName = GravitinoITUtils.genRandomName("tst_fs_conflict"); NameIdentifier filesetIdent = NameIdentifier.of(metalakeName, catalogName, schemaName, filesetName); @@ -405,6 +414,7 @@ public void testUpdateFilesetReturnsSuccessWhenVersionedMetaUpdateAffectsNoRows( AUDIT_INFO, "/tmp"); FilesetMetaService.getInstance().insertFileset(filesetEntity, true); + FilesetPO initialPO = getFilesetPO(filesetEntity.id()); AuditInfo conflictingAuditInfo = AuditInfo.builder() @@ -427,36 +437,26 @@ public void testUpdateFilesetReturnsSuccessWhenVersionedMetaUpdateAffectsNoRows( .build()) .build(); - Exception exception = - Assertions.assertThrows( - IOException.class, - () -> - FilesetMetaService.getInstance() - .updateFileset( - filesetIdent, - e -> { - // Simulate an optimistic locking conflict - try { - backend.update( - filesetIdent, - Entity.EntityType.FILESET, - entity -> { - FilesetEntity cloned = - createFilesetEntity( - entity.id(), - entity.namespace(), - entity.name(), - conflictingAuditInfo, - "/tmp"); - return cloned; - }); - } catch (Exception ex) { - throw new RuntimeException(ex); - } - return updatedFilesetEntity; - })); - Assertions.assertTrue( - exception.getMessage().contains("Failed to update the entity: " + filesetIdent)); + Assertions.assertThrows( + OptimisticLockException.class, + () -> + FilesetMetaService.getInstance() + .updateFileset( + filesetIdent, + e -> { + // Commit another alter after the outer call has read its snapshot. The + // outer write must then lose the current_version comparison. + updateFilesetUnchecked( + filesetIdent, + entity -> + createFilesetEntity( + entity.id(), + entity.namespace(), + entity.name(), + conflictingAuditInfo, + "/tmp")); + return updatedFilesetEntity; + })); FilesetEntity persistedEntity = FilesetMetaService.getInstance().getFilesetByIdentifier(filesetIdent); @@ -465,5 +465,574 @@ public void testUpdateFilesetReturnsSuccessWhenVersionedMetaUpdateAffectsNoRows( Assertions.assertNull(persistedEntity.properties()); Assertions.assertEquals("/tmp", persistedEntity.storageLocations().get(LOCATION_NAME_UNKNOWN)); Assertions.assertNotEquals(updatedFilesetEntity, persistedEntity); + FilesetPO currentPO = getFilesetPO(filesetEntity.id()); + Assertions.assertEquals( + initialPO.getCurrentVersion() + 1, currentPO.getCurrentVersion().longValue()); + Assertions.assertEquals(currentPO.getCurrentVersion(), currentPO.getLastVersion()); + Assertions.assertEquals(2, listFilesetVersions(filesetEntity.id()).size()); + } + + @TestTemplate + public void testOverwriteAdvancesVersionAndRejectsStaleAlter() throws IOException { + String filesetName = GravitinoITUtils.genRandomName("tst_fs_overwrite_occ"); + FilesetEntity original = + createFilesetEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofFileset(metalakeName, catalogName, schemaName), + filesetName, + AUDIT_INFO, + "/tmp-original"); + FilesetMetaService.getInstance().insertFileset(original, false); + FilesetPO beforeOverwrite = getFilesetPO(original.id()); + FilesetEntity replacement = + copyFileset( + original, + original.id(), + original.name(), + "overwrite winner", + "/tmp-overwrite", + original.auditInfo()); + + assertThrows( + OptimisticLockException.class, + () -> + FilesetMetaService.getInstance() + .updateFileset( + original.nameIdentifier(), + entity -> { + insertFilesetUnchecked(replacement, true); + FilesetEntity current = (FilesetEntity) entity; + return copyFileset( + current, + current.id(), + current.name(), + "stale alter", + "/tmp-stale", + current.auditInfo()); + })); + + FilesetEntity winner = + FilesetMetaService.getInstance().getFilesetByIdentifier(original.nameIdentifier()); + FilesetPO afterOverwrite = getFilesetPO(original.id()); + Assertions.assertEquals("overwrite winner", winner.comment()); + Assertions.assertEquals("/tmp-overwrite", winner.storageLocations().get(LOCATION_NAME_UNKNOWN)); + Assertions.assertEquals( + beforeOverwrite.getCurrentVersion() + 1, afterOverwrite.getCurrentVersion().longValue()); + Assertions.assertEquals(afterOverwrite.getCurrentVersion(), afterOverwrite.getLastVersion()); + } + + @TestTemplate + public void testNaturalKeyOverwriteUsesPersistedFilesetId() throws IOException { + String filesetName = GravitinoITUtils.genRandomName("tst_fs_natural_key_overwrite"); + FilesetEntity original = + createFilesetEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofFileset(metalakeName, catalogName, schemaName), + filesetName, + AUDIT_INFO, + "/tmp-original"); + FilesetMetaService.getInstance().insertFileset(original, false); + FilesetPO beforeOverwrite = getFilesetPO(original.id()); + Map replacementLocations = + ImmutableMap.of(LOCATION_NAME_UNKNOWN, "/tmp-replacement", "archive", "/tmp-archive"); + FilesetEntity replacement = + FilesetEntity.builder() + .withId(RandomIdGenerator.INSTANCE.nextId()) + .withName(original.name()) + .withNamespace(original.namespace()) + .withFilesetType(original.filesetType()) + .withStorageLocations(replacementLocations) + .withComment("replacement") + .withProperties(original.properties()) + .withAuditInfo(original.auditInfo()) + .build(); + + FilesetMetaService.getInstance().insertFileset(replacement, true); + + FilesetEntity stored = + FilesetMetaService.getInstance().getFilesetByIdentifier(original.nameIdentifier()); + FilesetPO afterOverwrite = getFilesetPO(original.id()); + Assertions.assertEquals(original.id(), stored.id()); + Assertions.assertEquals("replacement", stored.comment()); + Assertions.assertEquals(replacementLocations, stored.storageLocations()); + Assertions.assertEquals( + beforeOverwrite.getCurrentVersion() + 1, afterOverwrite.getCurrentVersion().longValue()); + } + + @TestTemplate + public void testAlterReportsNoSuchWhenRenamedConcurrently() throws IOException { + String filesetName = GravitinoITUtils.genRandomName("tst_fs_rename_conflict"); + FilesetEntity original = + createFilesetEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofFileset(metalakeName, catalogName, schemaName), + filesetName, + AUDIT_INFO, + "/tmp"); + FilesetMetaService.getInstance().insertFileset(original, false); + String renamedName = filesetName + "_winner"; + NameIdentifier renamedIdentifier = NameIdentifier.of(original.namespace(), renamedName); + + assertThrows( + NoSuchEntityException.class, + () -> + FilesetMetaService.getInstance() + .updateFileset( + original.nameIdentifier(), + entity -> { + updateFilesetUnchecked( + original.nameIdentifier(), + current -> + copyFileset( + current, + current.id(), + renamedName, + "rename winner", + "/tmp", + current.auditInfo())); + FilesetEntity current = (FilesetEntity) entity; + return copyFileset( + current, + current.id(), + current.name(), + "stale alter", + "/tmp", + current.auditInfo()); + })); + + assertThrows( + NoSuchEntityException.class, + () -> FilesetMetaService.getInstance().getFilesetByIdentifier(original.nameIdentifier())); + Assertions.assertEquals( + "rename winner", + FilesetMetaService.getInstance().getFilesetByIdentifier(renamedIdentifier).comment()); + } + + @TestTemplate + public void testDeleteRejectsStaleVersionAndKeepsVersions() throws IOException { + String filesetName = GravitinoITUtils.genRandomName("tst_fs_stale_delete"); + FilesetEntity original = + createFilesetEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofFileset(metalakeName, catalogName, schemaName), + filesetName, + AUDIT_INFO, + "/tmp-v1"); + FilesetMetaService.getInstance().insertFileset(original, false); + FilesetPO stalePO = getFilesetPO(original.id()); + + FilesetMetaService.getInstance() + .updateFileset( + original.nameIdentifier(), + entity -> { + FilesetEntity current = (FilesetEntity) entity; + return copyFileset( + current, + current.id(), + current.name(), + "winning alter", + "/tmp-v2", + current.auditInfo()); + }); + + assertThrows( + OptimisticLockException.class, + () -> + SessionUtils.doMultipleWithCommit( + () -> + FilesetMetaService.getInstance() + .deleteFilesetWithVersion(original.nameIdentifier(), stalePO))); + + FilesetEntity current = + FilesetMetaService.getInstance().getFilesetByIdentifier(original.nameIdentifier()); + Assertions.assertEquals("winning alter", current.comment()); + Assertions.assertEquals("/tmp-v2", current.storageLocations().get(LOCATION_NAME_UNKNOWN)); + Map versions = listFilesetVersions(original.id()); + Assertions.assertEquals(2, versions.size()); + assertVersionActive(versions, 1); + assertVersionActive(versions, 2); + } + + @TestTemplate + public void testDeleteReportsNoSuchWhenDeletedConcurrently() throws IOException { + String filesetName = GravitinoITUtils.genRandomName("tst_fs_double_delete"); + FilesetEntity fileset = + createFilesetEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofFileset(metalakeName, catalogName, schemaName), + filesetName, + AUDIT_INFO, + "/tmp"); + FilesetMetaService.getInstance().insertFileset(fileset, false); + FilesetPO stalePO = getFilesetPO(fileset.id()); + + FilesetMetaService.getInstance().deleteFileset(fileset.nameIdentifier()); + + assertThrows( + NoSuchEntityException.class, + () -> + SessionUtils.doMultipleWithCommit( + () -> + FilesetMetaService.getInstance() + .deleteFilesetWithVersion(fileset.nameIdentifier(), stalePO))); + } + + @TestTemplate + public void testDeleteAndGetReturnsSnapshotProtectedByDeleteCas() throws IOException { + String filesetName = GravitinoITUtils.genRandomName("tst_fs_delete_snapshot"); + FilesetEntity original = + createFilesetEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofFileset(metalakeName, catalogName, schemaName), + filesetName, + AUDIT_INFO, + "/tmp-v1"); + FilesetMetaService.getInstance().insertFileset(original, false); + FilesetEntity updated = + FilesetMetaService.getInstance() + .updateFileset( + original.nameIdentifier(), + entity -> { + FilesetEntity current = (FilesetEntity) entity; + return copyFileset( + current, + current.id(), + current.name(), + "snapshot selected by delete", + "/tmp-v2", + current.auditInfo()); + }); + + FilesetEntity deleted = + FilesetMetaService.getInstance().deleteFilesetAndGet(original.nameIdentifier()); + + Assertions.assertEquals(updated, deleted); + assertThrows( + NoSuchEntityException.class, + () -> FilesetMetaService.getInstance().getFilesetByIdentifier(original.nameIdentifier())); + Map versions = listFilesetVersions(original.id()); + assertVersionSoftDeleted(versions, 1); + assertVersionSoftDeleted(versions, 2); + } + + @TestTemplate + public void testUpdateRollsBackMetadataWhenVersionInsertFails() throws IOException { + String filesetName = GravitinoITUtils.genRandomName("tst_fs_update_rollback"); + FilesetEntity original = + createFilesetEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofFileset(metalakeName, catalogName, schemaName), + filesetName, + AUDIT_INFO, + "/tmp-v1"); + FilesetMetaService.getInstance().insertFileset(original, false); + FilesetPO initialPO = getFilesetPO(original.id()); + // fileset_meta carries no comment, so an over-long comment passes the metadata update and only + // fails once the version snapshot is written. + String tooLongComment = StringUtils.repeat("c", 300); + + // Each backend reports the rejected snapshot differently, so only the rollback below is + // asserted on. + assertThrows( + Exception.class, + () -> + FilesetMetaService.getInstance() + .updateFileset( + original.nameIdentifier(), + entity -> { + FilesetEntity current = (FilesetEntity) entity; + return copyFileset( + current, + current.id(), + current.name(), + tooLongComment, + "/tmp-v2", + current.auditInfo()); + })); + + FilesetEntity current = + FilesetMetaService.getInstance().getFilesetByIdentifier(original.nameIdentifier()); + FilesetPO currentPO = getFilesetPO(original.id()); + Assertions.assertEquals(original.comment(), current.comment()); + Assertions.assertEquals( + original.storageLocations().get(LOCATION_NAME_UNKNOWN), + current.storageLocations().get(LOCATION_NAME_UNKNOWN)); + Assertions.assertEquals(initialPO.getCurrentVersion(), currentPO.getCurrentVersion()); + Assertions.assertEquals(initialPO.getLastVersion(), currentPO.getLastVersion()); + } + + @TestTemplate + public void testAlterSkipsVersionsAlreadyStored() throws IOException { + String filesetName = GravitinoITUtils.genRandomName("tst_fs_stale_version"); + FilesetEntity original = + createFilesetEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofFileset(metalakeName, catalogName, schemaName), + filesetName, + AUDIT_INFO, + "/tmp-v1"); + FilesetMetaService.getInstance().insertFileset(original, false); + FilesetPO initialPO = getFilesetPO(original.id()); + + // A fileset written before the version reset was fixed owns snapshots above the version its + // metadata row records. The next alter has to start above those, not on top of them. + FilesetVersionPO staleVersion = + FilesetVersionPO.builder() + .withMetalakeId(initialPO.getMetalakeId()) + .withCatalogId(initialPO.getCatalogId()) + .withSchemaId(initialPO.getSchemaId()) + .withFilesetId(initialPO.getFilesetId()) + .withVersion(initialPO.getCurrentVersion() + 1) + .withFilesetComment("left behind by an older release") + // Use a different location name from the new snapshot. The legacy row therefore would + // not cause a unique-key collision; the metadata CAS itself must detect it. + .withLocationName("legacy-location") + .withStorageLocation("/tmp-stale") + .withDeletedAt(0L) + .build(); + SessionUtils.doWithCommit( + FilesetVersionMapper.class, mapper -> mapper.insertFilesetVersions(List.of(staleVersion))); + + FilesetEntity altered = + FilesetMetaService.getInstance() + .updateFileset( + original.nameIdentifier(), + entity -> { + FilesetEntity current = (FilesetEntity) entity; + return copyFileset( + current, + current.id(), + current.name(), + "altered past the stale version", + "/tmp-v2", + current.auditInfo()); + }); + + Assertions.assertEquals("altered past the stale version", altered.comment()); + FilesetPO afterAlter = getFilesetPO(original.id()); + Assertions.assertEquals( + staleVersion.getVersion() + 1, afterAlter.getCurrentVersion().longValue()); + FilesetEntity reloaded = + FilesetMetaService.getInstance().getFilesetByIdentifier(original.nameIdentifier()); + Assertions.assertEquals("altered past the stale version", reloaded.comment()); + Assertions.assertEquals("/tmp-v2", reloaded.storageLocations().get(LOCATION_NAME_UNKNOWN)); + } + + @TestTemplate + public void testOverwriteSkipsVersionsAlreadyStored() throws IOException { + String filesetName = GravitinoITUtils.genRandomName("tst_fs_overwrite_stale_version"); + FilesetEntity original = + createFilesetEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofFileset(metalakeName, catalogName, schemaName), + filesetName, + AUDIT_INFO, + "/tmp-v1"); + FilesetMetaService.getInstance().insertFileset(original, false); + FilesetPO initialPO = getFilesetPO(original.id()); + + // The same legacy shape the alter path already handles: a snapshot above the version the + // metadata row records. The overwrite derives its version from that row, so it has to be + // lifted above the snapshot instead of rewriting it. + FilesetVersionPO staleVersion = + FilesetVersionPO.builder() + .withMetalakeId(initialPO.getMetalakeId()) + .withCatalogId(initialPO.getCatalogId()) + .withSchemaId(initialPO.getSchemaId()) + .withFilesetId(initialPO.getFilesetId()) + .withVersion(initialPO.getCurrentVersion() + 1) + .withFilesetComment("left behind by an older release") + .withLocationName(LOCATION_NAME_UNKNOWN) + .withStorageLocation("/tmp-stale") + .withDeletedAt(0L) + .build(); + SessionUtils.doWithCommit( + FilesetVersionMapper.class, mapper -> mapper.insertFilesetVersions(List.of(staleVersion))); + + FilesetEntity replacement = + copyFileset( + original, + original.id(), + original.name(), + "overwritten past the stale version", + "/tmp-v2", + original.auditInfo()); + FilesetMetaService.getInstance().insertFileset(replacement, true); + + FilesetPO afterOverwrite = getFilesetPO(original.id()); + Assertions.assertEquals( + staleVersion.getVersion() + 1, afterOverwrite.getCurrentVersion().longValue()); + FilesetEntity stored = + FilesetMetaService.getInstance().getFilesetByIdentifier(original.nameIdentifier()); + Assertions.assertEquals("overwritten past the stale version", stored.comment()); + Assertions.assertEquals("/tmp-v2", stored.storageLocations().get(LOCATION_NAME_UNKNOWN)); + // The snapshot that was left behind is untouched at its own version. + Assertions.assertEquals( + "/tmp-stale", + storageLocationOfVersion(initialPO.getFilesetId(), staleVersion.getVersion())); + } + + @TestTemplate + public void testNaturalKeyOverwriteRewritesIdentifierProperty() throws IOException { + String filesetName = GravitinoITUtils.genRandomName("tst_fs_overwrite_identifier"); + FilesetEntity original = + createFilesetEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofFileset(metalakeName, catalogName, schemaName), + filesetName, + AUDIT_INFO, + "/tmp-original"); + FilesetMetaService.getInstance().insertFileset(original, false); + + long replacementId = RandomIdGenerator.INSTANCE.nextId(); + FilesetEntity replacement = + FilesetEntity.builder() + .withId(replacementId) + .withName(original.name()) + .withNamespace(original.namespace()) + .withFilesetType(original.filesetType()) + .withStorageLocations(ImmutableMap.of(LOCATION_NAME_UNKNOWN, "/tmp-replacement")) + .withComment("replacement") + .withProperties( + ImmutableMap.of( + StringIdentifier.ID_KEY, StringIdentifier.fromId(replacementId).toString())) + .withAuditInfo(original.auditInfo()) + .build(); + + FilesetMetaService.getInstance().insertFileset(replacement, true); + + // The overwrite keeps the fileset ID the database already had, so the identifier property has + // to name that ID as well instead of the one the rejected snapshot was built with. + FilesetEntity stored = + FilesetMetaService.getInstance().getFilesetByIdentifier(original.nameIdentifier()); + Assertions.assertEquals(original.id(), stored.id()); + Assertions.assertEquals( + StringIdentifier.fromId(original.id()).toString(), + stored.properties().get(StringIdentifier.ID_KEY)); + } + + @TestTemplate + public void testDeleteAndGetRefusesAPreCommitActionItCannotHonor() { + // Only the fileset path runs the action while the delete can still be rolled back. Any other + // entity type would run it after the commit, which is the opposite of what callers rely on, so + // it has to say so instead of doing it anyway. + Assertions.assertThrows( + UnsupportedOperationException.class, + () -> + backend.deleteAndGet( + NameIdentifier.of(metalakeName, catalogName, schemaName), + Entity.EntityType.SCHEMA, + SchemaEntity.class, + ignored -> { + throw new IllegalStateException("not reached"); + })); + } + + @TestTemplate + public void testAlterReportsConflictWhenTheNameWasTakenOver() throws IOException { + String filesetName = GravitinoITUtils.genRandomName("tst_fs_name_taken_over"); + FilesetEntity original = + createFilesetEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofFileset(metalakeName, catalogName, schemaName), + filesetName, + AUDIT_INFO, + "/tmp-original"); + FilesetMetaService.getInstance().insertFileset(original, false); + + // The fileset this alter resolved is renamed away and a different one takes over its name. The + // name still resolves, so the loser is told to retry rather than that the name is gone. + assertThrows( + OptimisticLockException.class, + () -> + FilesetMetaService.getInstance() + .updateFileset( + original.nameIdentifier(), + entity -> { + updateFilesetUnchecked( + original.nameIdentifier(), + current -> + copyFileset( + current, + current.id(), + filesetName + "_moved", + "rename winner", + "/tmp-moved", + current.auditInfo())); + insertFilesetUnchecked( + createFilesetEntity( + RandomIdGenerator.INSTANCE.nextId(), + original.namespace(), + filesetName, + AUDIT_INFO, + "/tmp-taken-over"), + false); + FilesetEntity current = (FilesetEntity) entity; + return copyFileset( + current, + current.id(), + current.name(), + "stale alter", + "/tmp-loser", + current.auditInfo()); + })); + } + + private String storageLocationOfVersion(Long filesetId, Long version) { + try (SqlSession sqlSession = + SqlSessionFactoryHelper.getInstance().getSqlSessionFactory().openSession(true); + Connection connection = sqlSession.getConnection(); + Statement statement = connection.createStatement(); + ResultSet rs = + statement.executeQuery( + String.format( + "SELECT storage_location FROM fileset_version_info" + + " WHERE fileset_id = %d AND version = %d AND deleted_at = 0", + filesetId, version))) { + return rs.next() ? rs.getString("storage_location") : null; + } catch (SQLException e) { + throw new RuntimeException("SQL execution failed", e); + } + } + + private FilesetPO getFilesetPO(Long filesetId) { + return SessionUtils.getWithoutCommit( + FilesetMetaMapper.class, mapper -> mapper.selectFilesetMetaById(filesetId)); + } + + private FilesetEntity copyFileset( + FilesetEntity source, + Long id, + String name, + String comment, + String location, + AuditInfo auditInfo) { + return FilesetEntity.builder() + .withId(id) + .withName(name) + .withNamespace(source.namespace()) + .withFilesetType(source.filesetType()) + .withStorageLocations(ImmutableMap.of(LOCATION_NAME_UNKNOWN, location)) + .withComment(comment) + .withProperties(source.properties()) + .withAuditInfo(auditInfo) + .build(); + } + + private void updateFilesetUnchecked( + NameIdentifier identifier, Function updater) { + try { + FilesetMetaService.getInstance().updateFileset(identifier, updater); + } catch (IOException e) { + throw new RuntimeException(e); + } + } + + private void insertFilesetUnchecked(FilesetEntity fileset, boolean overwrite) { + try { + FilesetMetaService.getInstance().insertFileset(fileset, overwrite); + } catch (IOException e) { + throw new RuntimeException(e); + } } } 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 994432dcaa5..2935e0b4e74 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 @@ -760,11 +760,8 @@ public void testUpdateFilesetPOVersion() throws JsonProcessingException { FilesetPO.builder().withMetalakeId(1L).withCatalogId(1L).withSchemaId(1L); FilesetPO initPO = POConverters.initializeFilesetPOWithVersion(filesetEntity, builder); - // map has updated - boolean checkNeedUpdate1 = - POConverters.checkFilesetVersionNeedUpdate(initPO.getFilesetVersionPOs(), updatedFileset); - FilesetPO updatePO1 = - POConverters.updateFilesetPOWithVersion(initPO, updatedFileset, checkNeedUpdate1); + // A content change advances the version and writes a complete new snapshot. + FilesetPO updatePO1 = POConverters.updateFilesetPOWithVersion(initPO, updatedFileset, null); assertEquals(1, initPO.getCurrentVersion()); assertEquals(1, initPO.getLastVersion()); assertEquals(0, initPO.getDeletedAt()); @@ -783,11 +780,9 @@ public void testUpdateFilesetPOVersion() throws JsonProcessingException { .readValue(updatePO1.getFilesetVersionPOs().get(0).getProperties(), Map.class); assertEquals("value1", updatedProperties.get("key")); - // will not update version, but update the fileset name - boolean checkNeedUpdate2 = - POConverters.checkFilesetVersionNeedUpdate(initPO.getFilesetVersionPOs(), updatedFileset1); - FilesetPO updatePO2 = - POConverters.updateFilesetPOWithVersion(initPO, updatedFileset1, checkNeedUpdate2); + // Metadata-only changes must also advance the OCC token. Reads join the version table through + // current_version, so the converter writes the unchanged content as a new complete snapshot. + FilesetPO updatePO2 = POConverters.updateFilesetPOWithVersion(initPO, updatedFileset1, null); Map storageLocations2 = updatePO2.getFilesetVersionPOs().stream() .collect( @@ -795,10 +790,17 @@ public void testUpdateFilesetPOVersion() throws JsonProcessingException { FilesetVersionPO::getLocationName, FilesetVersionPO::getStorageLocation)); assertEquals(filesetEntity.storageLocation(), storageLocations2.get(LOCATION_NAME_UNKNOWN)); assertEquals(filesetEntity.storageLocations(), storageLocations2); - assertEquals(1, updatePO2.getCurrentVersion()); - assertEquals(1, updatePO2.getLastVersion()); - assertEquals(1, updatePO2.getFilesetVersionPOs().get(0).getVersion()); + assertEquals(2, updatePO2.getCurrentVersion()); + assertEquals(2, updatePO2.getLastVersion()); + assertEquals(2, updatePO2.getFilesetVersionPOs().get(0).getVersion()); assertEquals("test1", updatePO2.getFilesetName()); + + // A snapshot stored above the version the metadata row records must not be rebuilt: the next + // version starts above every snapshot the fileset still owns. + FilesetPO updatePO3 = POConverters.updateFilesetPOWithVersion(initPO, updatedFileset, 7L); + assertEquals(8, updatePO3.getCurrentVersion()); + assertEquals(8, updatePO3.getLastVersion()); + assertEquals(8, updatePO3.getFilesetVersionPOs().get(0).getVersion()); } @Test