This is an automated email from the ASF dual-hosted git repository.
jerryshao pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new 15731007ef [#12650] improvement(core): add OCC for fileset writes
(#12656)
15731007ef is described below
commit 15731007efed9da8c840230d5539f22d76d77db1
Author: Qi Yu <[email protected]>
AuthorDate: Fri Aug 28 20:18:26 2026 +0800
[#12650] improvement(core): add OCC for fileset writes (#12656)
### What changes were proposed in this pull request?
- Use `current_version` as the OCC token for fileset alter and direct
delete.
- Advance the version and write a complete fileset snapshot for every
successful alter.
- Execute the root fileset CAS before version and dependent-row writes
in one transaction.
- Distinguish concurrent version changes from deleted, renamed, or moved
filesets.
- Keep versions monotonic during overwrite and use the persisted fileset
identity for version rows.
- Add comments explaining transaction ordering, CAS checks, and conflict
classification.
- Add service and SQL provider tests for the OCC behavior.
This is a subtask of #12166 and follows the table OCC implementation in
#12551.
### Why are the changes needed?
Concurrent fileset writes could overwrite the winning version metadata,
while a stale delete could remove data belonging to a newer fileset
version. Overwrite could also reset the OCC version.
Fix: #12650
### Does this PR introduce _any_ user-facing change?
Stale fileset writes now fail with the existing optimistic-lock or
not-found response instead of overwriting newer metadata. No API or
property is added.
### How was this patch tested?
- Ran the Fileset service, converter, and SQL provider tests on H2.
- Ran `:core:check -PskipITs -PskipDockerTests=true`.
- Ran Spotless and `git diff --check`.
- MySQL and PostgreSQL execution was attempted, but the local database
containers failed during connection setup before the test cases ran.
---
.../catalog/fileset/FilesetCatalogOperations.java | 103 ++--
.../fileset/TestFilesetCatalogOperations.java | 111 ++++
.../java/org/apache/gravitino/EntityStore.java | 79 +++
.../gravitino/storage/relational/JDBCBackend.java | 45 ++
.../storage/relational/RelationalBackend.java | 58 ++
.../storage/relational/RelationalEntityStore.java | 24 +-
.../relational/mapper/FilesetMetaMapper.java | 23 +-
.../mapper/FilesetMetaSQLProviderFactory.java | 24 +-
.../relational/mapper/FilesetVersionMapper.java | 9 +
.../mapper/FilesetVersionSQLProviderFactory.java | 10 +
.../provider/base/FilesetMetaBaseSQLProvider.java | 74 ++-
.../base/FilesetVersionBaseSQLProvider.java | 13 +
.../postgresql/FilesetMetaPostgreSQLProvider.java | 19 +-
.../relational/service/FilesetMetaService.java | 297 ++++++----
.../storage/relational/utils/POConverters.java | 94 ++-
.../base/TestFilesetMetaBaseSQLProvider.java | 77 +++
.../TestFilesetMetaPostgreSQLProvider.java | 51 ++
.../relational/service/TestFilesetMetaService.java | 633 +++++++++++++++++++--
.../storage/relational/utils/TestPOConverters.java | 28 +-
19 files changed, 1501 insertions(+), 271 deletions(-)
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 24670074f6..930429a2d0 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.collect.Maps;
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;
@@ -729,59 +730,75 @@ public class FilesetCatalogOperations extends
ManagedSchemaOperations
@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<IOException> exception = new AtomicReference<>();
- Map<String, Path> storageLocations =
- Maps.transformValues(filesetEntity.storageLocations(), Path::new);
- storageLocations.forEach(
- (locationName, location) -> {
- try {
- Map<String, String> 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<FilesetEntity> 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.
+ *
+ * <p>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<String, Path> storageLocations =
+ Maps.transformValues(filesetEntity.storageLocations(), Path::new);
+ for (Map.Entry<String, Path> entry : storageLocations.entrySet()) {
+ String locationName = entry.getKey();
+ Path location = entry.getValue();
+ Map<String, String> 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 d4b70d3a8d..1db30f01e2 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.commons.lang3.reflect.FieldUtils;
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.NoSuchEntityException;
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 @@ public class TestFilesetCatalogOperations {
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 1f844a2e27..c63458c720 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.io.IOException;
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 @@ public interface EntityStore extends Closeable {
*/
boolean delete(NameIdentifier ident, EntityType entityType, boolean cascade)
throws IOException;
+ /**
+ * The only post-delete action an implementation that cannot run it before
commit accepts.
+ *
+ * <p>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<? extends Entity> NO_POST_DELETE_ACTION = ignored -> {};
+
+ /**
+ * Returns the shared no-op post-delete action.
+ *
+ * @param <E> the entity type
+ * @return an action that does nothing
+ */
+ @SuppressWarnings("unchecked")
+ static <E extends Entity & HasIdentifier> Consumer<E> noPostDeleteAction() {
+ return (Consumer<E>) NO_POST_DELETE_ACTION;
+ }
+
+ /**
+ * Deletes an entity and returns the snapshot chosen by the delete operation.
+ *
+ * <p>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 <E> the entity type
+ * @return the deleted entity, or empty when it did not exist
+ * @throws IOException if the delete operation fails
+ */
+ default <E extends Entity & HasIdentifier> Optional<E> deleteAndGet(
+ NameIdentifier ident, EntityType entityType, Class<E> clazz) throws
IOException {
+ return deleteAndGet(ident, entityType, clazz, noPostDeleteAction());
+ }
+
+ /**
+ * Deletes an entity, runs an action against the deleted snapshot, and
returns that snapshot.
+ *
+ * <p>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 <E> the entity type
+ * @return the deleted entity, or empty when it did not exist
+ * @throws IOException if the delete operation fails
+ */
+ default <E extends Entity & HasIdentifier> Optional<E> deleteAndGet(
+ NameIdentifier ident, EntityType entityType, Class<E> clazz, Consumer<E>
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 11065934aa..2555396448 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.io.IOException;
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;
@@ -462,6 +464,49 @@ public class JDBCBackend implements RelationalBackend,
SupportsOrphanedRelationC
}
}
+ @Override
+ public <E extends Entity & HasIdentifier> Optional<E> deleteAndGet(
+ NameIdentifier ident,
+ Entity.EntityType entityType,
+ Class<E> clazz,
+ Consumer<E> 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 50e48af1c2..ea5a8aa46c 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 @@ package org.apache.gravitino.storage.relational;
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 @@ public interface RelationalBackend extends Closeable,
SupportsRelationOperations
boolean delete(NameIdentifier ident, Entity.EntityType entityType, boolean
cascade)
throws IOException;
+ /**
+ * Deletes an entity and returns the snapshot used by that delete.
+ *
+ * <p>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 <E> the concrete entity type
+ * @return the deleted entity, or empty when it did not exist
+ * @throws IOException if the store operation fails
+ */
+ default <E extends Entity & HasIdentifier> Optional<E> deleteAndGet(
+ NameIdentifier ident, Entity.EntityType entityType, Class<E> 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 <E> the concrete entity type
+ * @return the deleted entity, or empty when it did not exist
+ * @throws IOException if the store operation fails
+ */
+ default <E extends Entity & HasIdentifier> Optional<E> deleteAndGet(
+ NameIdentifier ident,
+ Entity.EntityType entityType,
+ Class<E> clazz,
+ Consumer<E> 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 bf6c50ee09..0d643dfba8 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.Arrays;
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 class RelationalEntityStore
public <E extends Entity & HasIdentifier> 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 class RelationalEntityStore
}
}
+ @Override
+ public <E extends Entity & HasIdentifier> Optional<E> deleteAndGet(
+ NameIdentifier ident,
+ Entity.EntityType entityType,
+ Class<E> clazz,
+ Consumer<E> postDeleteAction)
+ throws IOException {
+ try {
+ return backend.deleteAndGet(ident, entityType, clazz, postDeleteAction);
+ } finally {
+ cache.invalidate(ident, entityType);
+ }
+ }
+
@Override
public <R, E extends Exception> R executeInTransaction(Executable<R, E>
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 fcbbc66c0b..fbd2ca9c26 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 @@ public interface FilesetMetaMapper {
@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 @@ public interface FilesetMetaMapper {
method = "softDeleteFilesetMetasBySchemaIds")
Integer softDeleteFilesetMetasBySchemaIds(@Param("schemaIds") List<Long>
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 07aaebb1cb..b09dd069e8 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 class FilesetMetaSQLProviderFactory {
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 class FilesetMetaSQLProviderFactory {
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 931110faf0..2263b9f140 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 @@ public interface FilesetVersionMapper {
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 0d29d6f3f8..eebb7e6802 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 class FilesetVersionSQLProviderFactory {
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 8a1ad653c8..6a420fce1d 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 class FilesetMetaBaseSQLProvider {
+ " AND fm.deleted_at = 0 AND vi.deleted_at = 0";
}
+ /**
+ * Returns the active fileset metadata row selected by its natural key.
+ *
+ * <p>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 class FilesetMetaBaseSQLProvider {
+ " 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.
+ *
+ * <p>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 class FilesetMetaBaseSQLProvider {
+ " 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 class FilesetMetaBaseSQLProvider {
+ "</script>";
}
- 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 12204fe0ce..61e013ac1c 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 class FilesetVersionBaseSQLProvider {
+ " 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 ebb4d1731f..84e53bb79a 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 class FilesetMetaPostgreSQLProvider extends
FilesetMetaBaseSQLProvider {
}
@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 class FilesetMetaPostgreSQLProvider extends
FilesetMetaBaseSQLProvider {
+ " #{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 b29d198193..70ce93113a 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 static
org.apache.gravitino.metrics.source.MetricsSource.GRAVITINO_RELATI
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.HasIdentifier;
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 class FilesetMetaService {
fillFilesetPOBuilderParentEntityId(builder, filesetEntity.namespace());
FilesetPO po =
POConverters.initializeFilesetPOWithVersion(filesetEntity, builder);
+ AtomicReference<FilesetPO> 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 class FilesetMetaService {
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 class FilesetMetaService {
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.
+ *
+ * <p>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 class FilesetMetaService {
return POConverters.fromFilesetPOs(filesetPOs,
firstIdent.namespace());
});
}
+
+ /**
+ * Soft-deletes the observed fileset metadata row without starting a
transaction.
+ *
+ * <p>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<String, String> 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 e166a9f4c6..5557378eda 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.List;
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 class POConverters {
*
* @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<FilesetVersionPO> 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<FilesetVersionPO> 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 class POConverters {
.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 class POConverters {
}
}
- public static boolean checkFilesetVersionNeedUpdate(
- List<FilesetVersionPO> oldFilesetVersionPOs, FilesetEntity newFileset) {
- Map<String, String> 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<String, String> 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 0000000000..cdbfcffd37
--- /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 0000000000..49753bfdea
--- /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 677c6ada5a..5255065194 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.time.Instant;
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.EntityAlreadyExistsException;
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 @@ public class TestFilesetMetaService extends TestJDBCBackend
{
}
@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 class TestFilesetMetaService extends TestJDBCBackend
{
AUDIT_INFO,
"/tmp");
FilesetMetaService.getInstance().insertFileset(filesetEntity, true);
+ FilesetPO initialPO = getFilesetPO(filesetEntity.id());
AuditInfo conflictingAuditInfo =
AuditInfo.builder()
@@ -427,36 +437,26 @@ public class TestFilesetMetaService extends
TestJDBCBackend {
.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 class TestFilesetMetaService extends
TestJDBCBackend {
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<String, String> 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<Integer, Long> 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<Integer, Long> 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<FilesetEntity, FilesetEntity>
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 994432dcaa..2935e0b4e7 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 class TestPOConverters {
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 class TestPOConverters {
.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<String, String> storageLocations2 =
updatePO2.getFilesetVersionPOs().stream()
.collect(
@@ -795,10 +790,17 @@ public class TestPOConverters {
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
