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 c816482b8d [#14119] fix(fileset): Preserve metadata on cascade cleanup
failure (#14120)
c816482b8d is described below
commit c816482b8d24d4a77aaa32c87434398f2d030302
Author: mchades <[email protected]>
AuthorDate: Fri Oct 23 22:56:32 2026 +0800
[#14119] fix(fileset): Preserve metadata on cascade cleanup failure (#14120)
### What changes were proposed in this pull request?
Delete filesets through the existing single-fileset cleanup path before
deleting their schema. Stop on the first storage failure, preserve the
failed fileset and parent metadata, and clean each successfully deleted
fileset's secrets from its deleted snapshot.
Clean empty schema directories before the schema metadata transaction
commits. A schema-specific store operation deletes functions in that
same transaction, rejects any remaining fileset, and rolls back
schema/function metadata if directory cleanup fails.
### Why are the changes needed?
Catalog and metalake force-drop bypass the normal schema dispatcher.
Previously, this path removed metadata before cleaning managed fileset
storage, leaving failed cleanup without metadata for retry.
Schema-directory cleanup could also fail after schema metadata had
already committed.
Fix: #14119
### Does this PR introduce _any_ user-facing change?
Managed-fileset cleanup failures retain the failed fileset and its
parents. Empty schema-directory cleanup failures retain schema and
function metadata. Earlier successful fileset deletions remain
committed; already removed directories are accepted on retry. EXTERNAL
data and data in catalogs with filesystem operations disabled remain
untouched.
No REST API or configuration changes.
### How was this patch tested?
- `:catalogs:catalog-fileset:test -PskipITs`: 267 tests passed,
including schema-directory failure/rollback/retry, function
preservation, concurrent fileset creation, and partial multi-location
deletion with retry.
- `:core:coreH2Test --tests '*TestSchemaMetaService' --tests
'*TestFilesetMetaService' -PskipITs`: 53 tests passed, including
schema/function/changelog rollback and error propagation.
- `:core:coreUnitTest --tests '*TestCatalogManager' --tests
'*TestMetalakeManager' --tests '*TestSchemaOperationDispatcher' --tests
'*TestFilesetOperationDispatcher' -PskipITs`: 84 tests passed.
- `./gradlew spotlessApply` and `git diff --check` passed.
- Docker/MySQL/PostgreSQL and Python integration tests were not rerun
for this revision.
---------
Co-authored-by: Jerry Shao <[email protected]>
---
.../catalog/fileset/FilesetCatalogOperations.java | 245 ++++-------
.../fileset/TestFilesetCatalogOperations.java | 476 +++++++++++++++++++++
.../java/org/apache/gravitino/EntityStore.java | 21 +
.../gravitino/storage/relational/JDBCBackend.java | 82 ++--
.../storage/relational/RelationalBackend.java | 21 +
.../storage/relational/RelationalEntityStore.java | 14 +
.../relational/service/SchemaMetaService.java | 50 ++-
.../gravitino/catalog/TestCatalogManager.java | 36 +-
.../gravitino/metalake/TestMetalakeManager.java | 3 +
.../relational/service/TestSchemaMetaService.java | 167 ++++++++
10 files changed, 922 insertions(+), 193 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 1183684da7..2f8ef02dfd 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
@@ -60,7 +60,6 @@ import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
-import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Supplier;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
@@ -107,6 +106,7 @@ import
org.apache.gravitino.exceptions.NoSuchEntityException;
import org.apache.gravitino.exceptions.NoSuchFilesetException;
import org.apache.gravitino.exceptions.NoSuchLocationNameException;
import org.apache.gravitino.exceptions.NoSuchSchemaException;
+import org.apache.gravitino.exceptions.NonEmptyEntityException;
import org.apache.gravitino.exceptions.NonEmptySchemaException;
import org.apache.gravitino.exceptions.SchemaAlreadyExistsException;
import org.apache.gravitino.file.FileInfo;
@@ -745,38 +745,7 @@ public class FilesetCatalogOperations extends
ManagedSchemaOperations
@Override
public boolean dropFileset(NameIdentifier ident) {
- try {
- // 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);
- }
- }
- });
- return deletedFileset.isPresent();
- } catch (NoSuchEntityException ne) {
- LOG.warn("Fileset {} does not exist", ident);
- return false;
- } catch (UncheckedIOException uioe) {
- throw ExceptionMessages.wrap("Failed to delete fileset " + ident,
uioe.getCause());
- } catch (IOException ioe) {
- throw ExceptionMessages.wrap("Failed to delete fileset " + ident, ioe);
- }
+ return dropFilesetAndGet(ident).isPresent();
}
/**
@@ -925,135 +894,47 @@ public class FilesetCatalogOperations extends
ManagedSchemaOperations
throw new NonEmptySchemaException("Schema %s is not empty", ident);
}
- SchemaEntity schemaEntity = store.get(ident, Entity.EntityType.SCHEMA,
SchemaEntity.class);
- Map<String, String> properties =
-
Optional.ofNullable(schemaEntity.properties()).orElse(Collections.emptyMap());
- Map<String, Path> schemaPaths = getAndCheckSchemaPaths(ident.name(),
properties);
-
- boolean dropped = super.dropSchema(ident, cascade);
- if (disableFSOps) {
- // No FS cleanup; still remove fileset write-through secrets when
cascade deleted them.
- // Schema secrets are cleaned by SchemaOperationDispatcher.
- if (dropped && cascade) {
- for (FilesetEntity fileset : filesets) {
- secretManager.deleteSecretsFromProperties(fileset.properties());
- }
- }
- return dropped;
- }
-
- // If the schema entity is failed to be deleted, we should not delete
the storage location
- // and return false immediately.
- if (!dropped) {
- return false;
- }
-
- // Delete all the managed filesets no matter whether the storage
location is under the
- // schema path or not.
- // The reason why we delete the managed fileset's storage location one
by one is because we
- // may mis-delete the storage location of the external fileset if it
happens to be under
- // the schema path.
- ClassLoader cl = Thread.currentThread().getContextClassLoader();
- filesets.parallelStream()
- .filter(f -> f.filesetType() == Fileset.Type.MANAGED)
- .forEach(
- f -> {
- ClassLoader oldCl =
Thread.currentThread().getContextClassLoader();
- try {
- // parallelStream uses forkjoin thread pool, which has a
different classloader
- // than the catalog thread. We need to set the context
classloader to the
- // catalog's classloader to avoid classloading issues.
- Thread.currentThread().setContextClassLoader(cl);
- f.storageLocations()
- .forEach(
- (locationName, location) -> {
- try {
- Path filesetPath = new Path(location);
- Map<String, String> fsConf =
- mergeUpLevelConfigurations(ident,
f.properties(), filesetPath);
- FileSystem fs =
getFileSystemWithCache(filesetPath, fsConf);
- if (fs.exists(filesetPath)) {
- if (!fs.delete(filesetPath, true)) {
- LOG.warn(
- "Failed to delete fileset {} location:
{} with location name: {}",
- f.name(),
- filesetPath,
- locationName);
- }
- }
- } catch (IOException ioe) {
- LOG.warn(
- "Failed to delete fileset {} location: {}
with location name: {}",
- f.name(),
- location,
- locationName,
- ioe);
- }
- });
- } finally {
- Thread.currentThread().setContextClassLoader(oldCl);
- }
- });
-
- // Delete the schema path if it exists and is empty.
- if (!schemaPaths.isEmpty()) {
- AtomicReference<RuntimeException> exception = new AtomicReference<>();
- schemaPaths.forEach(
- (locationName, schemaPath) -> {
- try {
- Map<String, String> fsConf =
- mergeUpLevelConfigurations(ident,
schemaEntity.properties(), schemaPath);
- FileSystem fs = getFileSystemWithCache(schemaPath, fsConf);
- if (fs.exists(schemaPath)) {
- FileStatus[] statuses = fs.listStatus(schemaPath);
- if (statuses.length == 0) {
- if (fs.delete(schemaPath, true)) {
- LOG.info(
- "Deleted schema {} location {} with location name
{}",
- ident,
- schemaPath,
- locationName);
- } else {
- LOG.warn(
- "Failed to delete schema {} because it has
files/folders under location {} with location name {}",
- ident,
- schemaPath,
- locationName);
- }
- }
- }
- } catch (IOException ioe) {
- LOG.warn(
- "Failed to delete schema {} location {} with location name
{}",
- ident,
- schemaPath,
- locationName,
- ioe);
- exception.set(
- new RuntimeException("Failed to delete schema " + ident +
" location", ioe));
- }
- });
- if (exception.get() != null) {
- throw exception.get();
- }
- }
-
- // Cascade store.delete removes fileset entities without
- // FilesetOperationDispatcher.dropFileset;
- // clean fileset secrets after FS ops (still need plaintext for
mergeUpLevelConfigurations).
- // Schema secrets are cleaned by SchemaOperationDispatcher.
+ // Catalog force-drop reaches this method without
SchemaOperationDispatcher's fileset
+ // cleanup. Drop each fileset before its schema so a storage failure
preserves the failed
+ // fileset and its parent for retry. Stop at the first failure; earlier
drops stay committed.
if (cascade) {
for (FilesetEntity fileset : filesets) {
- secretManager.deleteSecretsFromProperties(fileset.properties());
+ dropFilesetAndGet(fileset.nameIdentifier())
+ .ifPresent(
+ deleted ->
secretManager.deleteSecretsFromProperties(deleted.properties()));
}
}
- LOG.info("Deleted schema {}", ident);
- return true;
+ // Only functions may be deleted with the schema; remaining filesets
still reject the drop.
+ // Keep function and schema metadata in one transaction until directory
cleanup succeeds.
+ boolean dropped =
+ store
+ .deleteSchemaAndGet(
+ ident,
+ cascade,
+ schemaEntity -> {
+ if (!disableFSOps) {
+ try {
+ deleteEmptySchemaLocations(ident, schemaEntity);
+ } catch (IOException ioe) {
+ throw new UncheckedIOException(ioe);
+ }
+ }
+ })
+ .isPresent();
+ if (dropped) {
+ LOG.info("Deleted schema {}", ident);
+ }
+ return dropped;
+ } catch (NonEmptyEntityException neee) {
+ throw new NonEmptySchemaException(neee, "Schema %s is not empty", ident);
} catch (NoSuchEntityException ne) {
LOG.warn("Schema {} does not exist", ident);
return false;
+ } catch (UncheckedIOException uioe) {
+ throw ExceptionMessages.wrap(
+ "Failed to delete schema " + ident + " location", uioe.getCause());
} catch (IOException ioe) {
throw ExceptionMessages.wrap("Failed to delete schema " + ident + "
location", ioe);
}
@@ -1126,6 +1007,64 @@ public class FilesetCatalogOperations extends
ManagedSchemaOperations
}
}
+ private void deleteEmptySchemaLocations(NameIdentifier ident, SchemaEntity
schemaEntity)
+ throws IOException {
+ Map<String, String> properties =
+
Optional.ofNullable(schemaEntity.properties()).orElse(Collections.emptyMap());
+ Map<String, Path> schemaPaths = getAndCheckSchemaPaths(ident.name(),
properties);
+ for (Map.Entry<String, Path> entry : schemaPaths.entrySet()) {
+ String locationName = entry.getKey();
+ Path schemaPath = entry.getValue();
+ Map<String, String> fsConf = mergeUpLevelConfigurations(ident,
properties, schemaPath);
+ FileSystem fs = getFileSystemWithCache(schemaPath, fsConf);
+ // Filesystem changes cannot be rolled back with metadata. A retry may
find a location gone.
+ if (!fs.exists(schemaPath) || fs.listStatus(schemaPath).length != 0) {
+ continue;
+ }
+ if (!fs.delete(schemaPath, true) && fs.exists(schemaPath)) {
+ throw new IOException(
+ String.format(
+ "Failed to delete schema %s location %s with location name %s",
+ ident, schemaPath, locationName));
+ }
+ LOG.info(
+ "Deleted schema {} location {} with location name {}", ident,
schemaPath, locationName);
+ }
+ }
+
+ private Optional<FilesetEntity> dropFilesetAndGet(NameIdentifier ident) {
+ try {
+ // 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.
+ return 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);
+ }
+ }
+ });
+ } catch (NoSuchEntityException ne) {
+ LOG.warn("Fileset {} does not exist", ident);
+ return Optional.empty();
+ } catch (UncheckedIOException uioe) {
+ throw ExceptionMessages.wrap("Failed to delete fileset " + ident,
uioe.getCause());
+ } catch (IOException ioe) {
+ throw ExceptionMessages.wrap("Failed to delete fileset " + ident, ioe);
+ }
+ }
+
private void validateLocationHierarchy(
Map<String, String> schemaLocations, Map<String, String>
filesetLocations) {
if (schemaLocations == null
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 0c7f1c33fe..644cf7c9d2 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
@@ -66,6 +66,7 @@ import java.net.ConnectException;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.time.Instant;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
@@ -105,6 +106,7 @@ import org.apache.gravitino.audit.CallerContext;
import org.apache.gravitino.audit.FilesetAuditConstants;
import org.apache.gravitino.audit.FilesetDataOperation;
import org.apache.gravitino.cache.NoOpsCache;
+import org.apache.gravitino.catalog.ManagedFunctionOperations;
import org.apache.gravitino.catalog.hadoop.fs.FileSystemProvider;
import org.apache.gravitino.catalog.hadoop.fs.FileSystemUtils;
import org.apache.gravitino.catalog.hadoop.fs.LocalFileSystemProvider;
@@ -128,11 +130,17 @@ import
org.apache.gravitino.exceptions.SchemaAlreadyExistsException;
import org.apache.gravitino.file.FileInfo;
import org.apache.gravitino.file.Fileset;
import org.apache.gravitino.file.FilesetChange;
+import org.apache.gravitino.function.FunctionDefinitions;
+import org.apache.gravitino.function.FunctionImpl;
+import org.apache.gravitino.function.FunctionImpls;
+import org.apache.gravitino.function.FunctionParam;
+import org.apache.gravitino.function.FunctionType;
import org.apache.gravitino.meta.AuditInfo;
import org.apache.gravitino.meta.BaseMetalake;
import org.apache.gravitino.meta.CatalogEntity;
import org.apache.gravitino.meta.FilesetEntity;
import org.apache.gravitino.meta.SchemaVersion;
+import org.apache.gravitino.rel.types.Types;
import org.apache.gravitino.secret.SecretConstants;
import org.apache.gravitino.secret.SecretManager;
import org.apache.gravitino.secret.SecretMaterial;
@@ -150,6 +158,7 @@ import
org.apache.gravitino.storage.relational.service.SchemaMetaService;
import org.apache.gravitino.utils.NameIdentifierUtil;
import org.apache.gravitino.utils.PrincipalUtils;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocalFileSystem;
import org.apache.hadoop.fs.Path;
@@ -869,6 +878,457 @@ public class TestFilesetCatalogOperations {
}
}
+ @ParameterizedTest
+ @CsvSource({"false, false", "false, true", "true, false", "true, true"})
+ void testDropSchemaPreservesMetadataOnDirectoryCleanupFailure(
+ boolean cascade, boolean throwException) throws IOException {
+ String schemaName = "schema_directory_failure_" + generateTestId();
+ String catalogPath = TEST_ROOT_PATH + "/" + schemaName;
+ NameIdentifier schemaIdent = NameIdentifier.of("m1", "c1", schemaName);
+ NameIdentifier functionIdent = NameIdentifier.of("m1", "c1", schemaName,
"function");
+ Path schemaPath = new Path(catalogPath, schemaName);
+ long schemaId = generateTestId();
+ SecretUrn urn = writeThroughUrn("schema", schemaId, "schema-sk");
+ secretManager.writeSecrets(List.of(new SecretMaterial(urn,
"schema-secret")));
+
+ FileSystem realFileSystem = schemaPath.getFileSystem(new Configuration());
+ FileSystem failingFileSystem = Mockito.mock(FileSystem.class);
+ when(failingFileSystem.exists(schemaPath)).thenReturn(true);
+ when(failingFileSystem.listStatus(schemaPath)).thenReturn(new
FileStatus[0]);
+ if (throwException) {
+ when(failingFileSystem.delete(schemaPath, true))
+ .thenThrow(new IOException("permission denied"));
+ } else {
+ when(failingFileSystem.delete(schemaPath, true)).thenReturn(false);
+ }
+
+ try (FilesetCatalogOperations ops =
+ Mockito.spy(new FilesetCatalogOperations(store, secretManager))) {
+ ops.initialize(
+ ImmutableMap.of(LOCATION, catalogPath),
+ randomCatalogInfo("m1", "c1"),
+ FILESET_PROPERTIES_METADATA);
+ Schema schema =
+ ops.createSchema(
+ schemaIdent,
+ "comment",
+ StringIdentifier.newPropertiesWithId(
+ StringIdentifier.fromId(schemaId), Map.of("schema-sk",
urn.toString())));
+ if (cascade) {
+ createFunction(functionIdent);
+ }
+ doReturn(failingFileSystem)
+ .when(ops)
+ .getFileSystemWithCache(Mockito.eq(schemaPath), any(Map.class));
+ Mockito.clearInvocations(ops);
+
+ Assertions.assertThrows(RuntimeException.class, () ->
ops.dropSchema(schemaIdent, cascade));
+ Assertions.assertTrue(store.exists(schemaIdent,
Entity.EntityType.SCHEMA));
+ Assertions.assertEquals(cascade, store.exists(functionIdent,
Entity.EntityType.FUNCTION));
+ Assertions.assertEquals(schema.properties(),
ops.loadSchema(schemaIdent).properties());
+ Assertions.assertTrue(realFileSystem.exists(schemaPath));
+ Assertions.assertEquals("schema-secret", secretManager.readSecret(urn));
+ Mockito.verify(ops)
+ .getFileSystemWithCache(
+ Mockito.eq(schemaPath),
+ Mockito.argThat(fsConf ->
"schema-secret".equals(fsConf.get("schema-sk"))));
+
+ Mockito.doCallRealMethod()
+ .when(ops)
+ .getFileSystemWithCache(Mockito.eq(schemaPath), any(Map.class));
+ Assertions.assertTrue(ops.dropSchema(schemaIdent, cascade));
+ Assertions.assertFalse(store.exists(schemaIdent,
Entity.EntityType.SCHEMA));
+ Assertions.assertFalse(store.exists(functionIdent,
Entity.EntityType.FUNCTION));
+ Assertions.assertFalse(realFileSystem.exists(schemaPath));
+ // Schema secrets remain available until the caller handles the
successful return.
+ Assertions.assertEquals("schema-secret", secretManager.readSecret(urn));
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testDropSchemaWhenDirectoryDisappearsDuringDelete(boolean
throwException)
+ throws IOException {
+ String schemaName = "schema_directory_disappeared_" + generateTestId();
+ String catalogPath = TEST_ROOT_PATH + "/" + schemaName;
+ createSchema(schemaName, "comment", catalogPath, null);
+ NameIdentifier schemaIdent = NameIdentifier.of("m1", "c1", schemaName);
+ NameIdentifier functionIdent = NameIdentifier.of("m1", "c1", schemaName,
"function");
+ createFunction(functionIdent);
+ Path schemaPath = new Path(catalogPath, schemaName);
+ FileSystem realFileSystem = schemaPath.getFileSystem(new Configuration());
+ FileSystem failingFileSystem = Mockito.mock(FileSystem.class);
+ when(failingFileSystem.exists(schemaPath))
+ .thenAnswer(invocation -> realFileSystem.exists(schemaPath));
+ when(failingFileSystem.listStatus(schemaPath)).thenReturn(new
FileStatus[0]);
+ when(failingFileSystem.delete(schemaPath, true))
+ .thenAnswer(
+ invocation -> {
+ Assertions.assertTrue(realFileSystem.delete(schemaPath, true));
+ if (throwException) {
+ throw new IOException("delete response lost");
+ }
+ return false;
+ });
+
+ 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(Mockito.eq(schemaPath), any(Map.class));
+
+ if (throwException) {
+ Assertions.assertThrows(RuntimeException.class, () ->
ops.dropSchema(schemaIdent, true));
+ Assertions.assertTrue(store.exists(schemaIdent,
Entity.EntityType.SCHEMA));
+ Assertions.assertTrue(store.exists(functionIdent,
Entity.EntityType.FUNCTION));
+ Assertions.assertFalse(realFileSystem.exists(schemaPath));
+ Mockito.doCallRealMethod()
+ .when(ops)
+ .getFileSystemWithCache(Mockito.eq(schemaPath), any(Map.class));
+ }
+ Assertions.assertTrue(ops.dropSchema(schemaIdent, true));
+ Assertions.assertFalse(store.exists(schemaIdent,
Entity.EntityType.SCHEMA));
+ Assertions.assertFalse(store.exists(functionIdent,
Entity.EntityType.FUNCTION));
+ Assertions.assertFalse(realFileSystem.exists(schemaPath));
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testDropSchemaWithFunctions(boolean withFileset) throws IOException {
+ String schemaName = "schema_cascade_functions_" + generateTestId();
+ String catalogPath = TEST_ROOT_PATH + "/" + schemaName;
+ createSchema(schemaName, "comment", catalogPath, null);
+ NameIdentifier schemaIdent = NameIdentifier.of("m1", "c1", schemaName);
+ NameIdentifier functionIdent = NameIdentifier.of("m1", "c1", schemaName,
"function");
+ createFunction(functionIdent);
+ if (withFileset) {
+ createFileset("fileset", schemaName, "comment", Fileset.Type.MANAGED,
catalogPath, null);
+ }
+ NameIdentifier filesetIdent = NameIdentifier.of("m1", "c1", schemaName,
"fileset");
+ Path schemaPath = new Path(catalogPath, schemaName);
+ FileSystem fs = schemaPath.getFileSystem(new Configuration());
+
+ try (FilesetCatalogOperations ops = new FilesetCatalogOperations(store,
secretManager)) {
+ ops.initialize(
+ ImmutableMap.of(LOCATION, catalogPath),
+ randomCatalogInfo("m1", "c1"),
+ FILESET_PROPERTIES_METADATA);
+ Assertions.assertThrows(
+ NonEmptySchemaException.class, () -> ops.dropSchema(schemaIdent,
false));
+ Assertions.assertTrue(store.exists(schemaIdent,
Entity.EntityType.SCHEMA));
+ Assertions.assertTrue(store.exists(functionIdent,
Entity.EntityType.FUNCTION));
+ Assertions.assertEquals(withFileset, store.exists(filesetIdent,
Entity.EntityType.FILESET));
+ Assertions.assertTrue(fs.exists(schemaPath));
+
+ Assertions.assertTrue(ops.dropSchema(schemaIdent, true));
+ Assertions.assertFalse(store.exists(schemaIdent,
Entity.EntityType.SCHEMA));
+ Assertions.assertFalse(store.exists(functionIdent,
Entity.EntityType.FUNCTION));
+ Assertions.assertFalse(store.exists(filesetIdent,
Entity.EntityType.FILESET));
+ Assertions.assertFalse(fs.exists(schemaPath));
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testDropSchemaStopsOnStorageFailureAndCanRetry(boolean throwException)
throws IOException {
+ String schemaName = "schema_cascade_failure_" + generateTestId();
+ String catalogPath = TEST_ROOT_PATH + "/" + schemaName;
+ createSchema(schemaName, "comment", catalogPath, null);
+ NameIdentifier schemaIdent = NameIdentifier.of("m1", "c1", schemaName);
+ NameIdentifier functionIdent = NameIdentifier.of("m1", "c1", schemaName,
"function");
+ createFunction(functionIdent);
+ Namespace filesetNs = Namespace.of("m1", "c1", schemaName);
+ List<FilesetEntity> filesets = new ArrayList<>();
+ List<SecretUrn> secrets = new ArrayList<>();
+ for (String name : List.of("first", "failed", "last")) {
+ SecretUrn urn =
+ SecretUrn.buildWriteThrough(
+ "memory",
+ Map.of(
+ SecretConstants.ATTR_ENTITY_TYPE, "fileset",
+ SecretConstants.ATTR_ENTITY_ID,
String.valueOf(generateTestId()),
+ SecretConstants.ATTR_PROPERTY_KEY, "test-secret"));
+ secretManager.writeSecrets(List.of(new SecretMaterial(urn,
"test-value")));
+ secrets.add(urn);
+ createFileset(
+ name,
+ schemaName,
+ "comment",
+ Fileset.Type.MANAGED,
+ catalogPath,
+ null,
+ Map.of("test-secret", urn.toString()));
+ filesets.add(
+ store.get(
+ NameIdentifier.of(filesetNs, name), Entity.EntityType.FILESET,
FilesetEntity.class));
+ }
+
+ EntityStore orderedStore = Mockito.spy(store);
+ doReturn(filesets)
+ .when(orderedStore)
+ .list(filesetNs, FilesetEntity.class, Entity.EntityType.FILESET);
+ Path failedPath = new Path(filesets.get(1).storageLocation());
+ FileSystem realFileSystem = failedPath.getFileSystem(new Configuration());
+ FileSystem failingFileSystem = Mockito.mock(FileSystem.class);
+ when(failingFileSystem.exists(failedPath)).thenReturn(true);
+ if (throwException) {
+ when(failingFileSystem.delete(failedPath, true))
+ .thenThrow(new IOException("permission denied"));
+ } else {
+ when(failingFileSystem.delete(failedPath, true)).thenReturn(false);
+ }
+
+ try (FilesetCatalogOperations ops =
+ Mockito.spy(new FilesetCatalogOperations(orderedStore,
secretManager))) {
+ ops.initialize(
+ ImmutableMap.of(LOCATION, catalogPath),
+ randomCatalogInfo("m1", "c1"),
+ FILESET_PROPERTIES_METADATA);
+ doReturn(failingFileSystem)
+ .when(ops)
+ .getFileSystemWithCache(Mockito.eq(failedPath), any(Map.class));
+
+ Assertions.assertThrows(RuntimeException.class, () ->
ops.dropSchema(schemaIdent, true));
+ Assertions.assertTrue(store.exists(schemaIdent,
Entity.EntityType.SCHEMA));
+ Assertions.assertTrue(store.exists(functionIdent,
Entity.EntityType.FUNCTION));
+ for (int i = 0; i < filesets.size(); i++) {
+ FilesetEntity fileset = filesets.get(i);
+ Path path = new Path(fileset.storageLocation());
+ Assertions.assertEquals(
+ i != 0, store.exists(fileset.nameIdentifier(),
Entity.EntityType.FILESET));
+ Assertions.assertEquals(i != 0, realFileSystem.exists(path));
+ }
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () ->
secretManager.readSecret(secrets.get(0)));
+ Assertions.assertEquals("test-value",
secretManager.readSecret(secrets.get(1)));
+ Assertions.assertEquals("test-value",
secretManager.readSecret(secrets.get(2)));
+
+ // Retry with the real filesystem after repairing access to the failed
location.
+ Mockito.doCallRealMethod()
+ .when(ops)
+ .getFileSystemWithCache(Mockito.eq(failedPath), any(Map.class));
+ Assertions.assertTrue(ops.dropSchema(schemaIdent, true));
+ Assertions.assertFalse(store.exists(schemaIdent,
Entity.EntityType.SCHEMA));
+ Assertions.assertFalse(store.exists(functionIdent,
Entity.EntityType.FUNCTION));
+ for (FilesetEntity fileset : filesets) {
+ Assertions.assertFalse(store.exists(fileset.nameIdentifier(),
Entity.EntityType.FILESET));
+ Path path = new Path(fileset.storageLocation());
+ Assertions.assertFalse(realFileSystem.exists(path));
+ }
+ for (SecretUrn urn : secrets) {
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () ->
secretManager.readSecret(urn));
+ }
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testDropSchemaRetriesPartiallyDeletedFilesetLocations(boolean
throwException)
+ throws IOException {
+ String schemaName = "schema_cascade_partial_locations_" + generateTestId();
+ String catalogPath = TEST_ROOT_PATH + "/" + schemaName;
+ createSchema(schemaName, "comment", catalogPath, null);
+ NameIdentifier schemaIdent = NameIdentifier.of("m1", "c1", schemaName);
+ NameIdentifier filesetIdent = NameIdentifier.of("m1", "c1", schemaName,
"fileset");
+ SecretUrn urn = writeThroughUrn("fileset", generateTestId(),
"test-secret");
+ secretManager.writeSecrets(List.of(new SecretMaterial(urn, "test-value")));
+ Map<String, String> catalogProperties = ImmutableMap.of(LOCATION,
catalogPath);
+ Fileset fileset =
+ createMultiLocationFileset(
+ filesetIdent.name(),
+ schemaName,
+ "comment",
+ Fileset.Type.MANAGED,
+ catalogProperties,
+ ImmutableMap.of(
+ LOCATION_NAME_UNKNOWN,
+ catalogPath + "/" + schemaName + "/primary",
+ "secondary",
+ catalogPath + "/" + schemaName + "/secondary"),
+ ImmutableMap.of(
+ PROPERTY_DEFAULT_LOCATION_NAME,
+ LOCATION_NAME_UNKNOWN,
+ "test-secret",
+ urn.toString()));
+ List<Path> locations =
+
fileset.storageLocations().values().stream().map(Path::new).collect(Collectors.toList());
+ Assertions.assertEquals(2, locations.size());
+ Assertions.assertEquals(2, Set.copyOf(locations).size());
+
+ try (FileSystem realFileSystem = FileSystem.newInstance(new
Configuration());
+ FilesetCatalogOperations ops =
+ Mockito.spy(new FilesetCatalogOperations(store, secretManager))) {
+ for (Path location : locations) {
+ Assertions.assertTrue(realFileSystem.exists(location));
+ }
+ List<Path> deleteAttempts = new ArrayList<>();
+ FileSystem failingFileSystem = Mockito.mock(FileSystem.class);
+ when(failingFileSystem.exists(any(Path.class)))
+ .thenAnswer(invocation ->
realFileSystem.exists(invocation.getArgument(0)));
+ when(failingFileSystem.delete(any(Path.class), Mockito.eq(true)))
+ .thenAnswer(
+ invocation -> {
+ Path location = invocation.getArgument(0);
+ deleteAttempts.add(location);
+ // Fail the second actual delete without relying on the
location map's order.
+ if (deleteAttempts.size() == 2) {
+ if (throwException) {
+ throw new IOException("permission denied");
+ }
+ return false;
+ }
+ return realFileSystem.delete(location, true);
+ });
+ ops.initialize(catalogProperties, randomCatalogInfo("m1", "c1"),
FILESET_PROPERTIES_METADATA);
+
doReturn(failingFileSystem).when(ops).getFileSystemWithCache(any(Path.class),
any(Map.class));
+
+ Assertions.assertThrows(RuntimeException.class, () ->
ops.dropSchema(schemaIdent, true));
+ Assertions.assertEquals(2, deleteAttempts.size());
+ Assertions.assertEquals(Set.copyOf(locations),
Set.copyOf(deleteAttempts));
+ Assertions.assertFalse(realFileSystem.exists(deleteAttempts.get(0)));
+ Assertions.assertTrue(realFileSystem.exists(deleteAttempts.get(1)));
+ Assertions.assertTrue(store.exists(schemaIdent,
Entity.EntityType.SCHEMA));
+ FilesetEntity survivingFileset =
+ store.get(filesetIdent, Entity.EntityType.FILESET,
FilesetEntity.class);
+ Assertions.assertEquals(fileset.storageLocations(),
survivingFileset.storageLocations());
+ Assertions.assertEquals(fileset.properties(),
survivingFileset.properties());
+ Assertions.assertEquals("test-value", secretManager.readSecret(urn));
+
+ // Retry without recreating the first location, which was already
removed before the failure.
+
Mockito.doCallRealMethod().when(ops).getFileSystemWithCache(any(Path.class),
any(Map.class));
+ Assertions.assertTrue(ops.dropSchema(schemaIdent, true));
+ Assertions.assertFalse(store.exists(filesetIdent,
Entity.EntityType.FILESET));
+ Assertions.assertFalse(store.exists(schemaIdent,
Entity.EntityType.SCHEMA));
+ for (Path location : locations) {
+ Assertions.assertFalse(realFileSystem.exists(location));
+ }
+ Assertions.assertThrows(IllegalArgumentException.class, () ->
secretManager.readSecret(urn));
+ }
+ }
+
+ @Test
+ void testDropSchemaRejectsFilesetCreatedAfterListing() throws IOException {
+ String schemaName = "schema_cascade_new_fileset_" + generateTestId();
+ String catalogPath = TEST_ROOT_PATH + "/" + schemaName;
+ createSchema(schemaName, "comment", catalogPath, null);
+ NameIdentifier schemaIdent = NameIdentifier.of("m1", "c1", schemaName);
+ NameIdentifier functionIdent = NameIdentifier.of("m1", "c1", schemaName,
"function");
+ createFunction(functionIdent);
+ Namespace filesetNs = Namespace.of("m1", "c1", schemaName);
+ Fileset first =
+ createFileset("first", schemaName, "comment", Fileset.Type.MANAGED,
catalogPath, null);
+ NameIdentifier remainingIdent = NameIdentifier.of(filesetNs, "remaining");
+ Path remainingPath = new Path(new Path(catalogPath, schemaName),
remainingIdent.name());
+ FileSystem fs = remainingPath.getFileSystem(new Configuration());
+ long remainingId = generateTestId();
+ SecretUrn urn =
+ SecretUrn.buildWriteThrough(
+ "memory",
+ Map.of(
+ SecretConstants.ATTR_ENTITY_TYPE, "fileset",
+ SecretConstants.ATTR_ENTITY_ID, String.valueOf(remainingId),
+ SecretConstants.ATTR_PROPERTY_KEY, "test-secret"));
+ secretManager.writeSecrets(List.of(new SecretMaterial(urn, "test-value")));
+ Map<String, String> remainingProperties =
+ StringIdentifier.newPropertiesWithId(
+ StringIdentifier.fromId(remainingId), Map.of("test-secret",
urn.toString()));
+
+ EntityStore interleavedStore = Mockito.spy(store);
+ try (FilesetCatalogOperations ops =
+ new FilesetCatalogOperations(interleavedStore, secretManager)) {
+ ops.initialize(
+ ImmutableMap.of(LOCATION, catalogPath),
+ randomCatalogInfo("m1", "c1"),
+ FILESET_PROPERTIES_METADATA);
+ Mockito.doAnswer(
+ invocation -> {
+ List<FilesetEntity> listed =
+ store.list(filesetNs, FilesetEntity.class,
Entity.EntityType.FILESET);
+ // Simulate another writer creating a fileset after the
deletion snapshot is read.
+ ops.createFileset(
+ remainingIdent,
+ "comment",
+ Fileset.Type.MANAGED,
+ remainingPath.toString(),
+ remainingProperties);
+ return listed;
+ })
+ .when(interleavedStore)
+ .list(filesetNs, FilesetEntity.class, Entity.EntityType.FILESET);
+
+ Assertions.assertThrows(
+ NonEmptySchemaException.class, () -> ops.dropSchema(schemaIdent,
true));
+ Assertions.assertTrue(store.exists(schemaIdent,
Entity.EntityType.SCHEMA));
+ Assertions.assertTrue(store.exists(remainingIdent,
Entity.EntityType.FILESET));
+ Assertions.assertTrue(store.exists(functionIdent,
Entity.EntityType.FUNCTION));
+ Assertions.assertTrue(fs.exists(remainingPath));
+ Assertions.assertEquals("test-value", secretManager.readSecret(urn));
+ Assertions.assertFalse(
+ store.exists(NameIdentifier.of(filesetNs, first.name()),
Entity.EntityType.FILESET));
+ Assertions.assertFalse(fs.exists(new Path(first.storageLocation())));
+
+ Mockito.doCallRealMethod()
+ .when(interleavedStore)
+ .list(filesetNs, FilesetEntity.class, Entity.EntityType.FILESET);
+ Assertions.assertTrue(ops.dropSchema(schemaIdent, true));
+ Assertions.assertFalse(store.exists(schemaIdent,
Entity.EntityType.SCHEMA));
+ Assertions.assertFalse(store.exists(remainingIdent,
Entity.EntityType.FILESET));
+ Assertions.assertFalse(store.exists(functionIdent,
Entity.EntityType.FUNCTION));
+ Assertions.assertFalse(fs.exists(remainingPath));
+ Assertions.assertThrows(IllegalArgumentException.class, () ->
secretManager.readSecret(urn));
+ }
+ }
+
+ @ParameterizedTest
+ @CsvSource({"EXTERNAL, false", "MANAGED, true"})
+ void testDropSchemaCleansSecretsWithoutDeletingStorage(Fileset.Type type,
boolean disableFSOps)
+ throws IOException {
+ String schemaName = "schema_cascade_metadata_only_" + generateTestId();
+ String catalogPath = TEST_ROOT_PATH + "/" + schemaName;
+ createSchema(schemaName, "comment", catalogPath, null);
+ NameIdentifier schemaIdent = NameIdentifier.of("m1", "c1", schemaName);
+ NameIdentifier filesetIdent = NameIdentifier.of("m1", "c1", schemaName,
"fileset");
+ SecretUrn urn =
+ SecretUrn.buildWriteThrough(
+ "memory",
+ Map.of(
+ SecretConstants.ATTR_ENTITY_TYPE, "fileset",
+ SecretConstants.ATTR_ENTITY_ID,
String.valueOf(generateTestId()),
+ SecretConstants.ATTR_PROPERTY_KEY, "test-secret"));
+ secretManager.writeSecrets(List.of(new SecretMaterial(urn, "test-value")));
+ Fileset fileset =
+ createFileset(
+ filesetIdent.name(),
+ schemaName,
+ "comment",
+ type,
+ catalogPath,
+ catalogPath + "/fileset",
+ Map.of("test-secret", urn.toString()));
+ Path path = new Path(fileset.storageLocation());
+ FileSystem fs = path.getFileSystem(new Configuration());
+ Assertions.assertTrue(fs.exists(path));
+
+ try (FilesetCatalogOperations ops = new FilesetCatalogOperations(store,
secretManager)) {
+ ops.initialize(
+ ImmutableMap.of(
+ LOCATION, catalogPath, DISABLE_FILESYSTEM_OPS,
String.valueOf(disableFSOps)),
+ randomCatalogInfo("m1", "c1"),
+ FILESET_PROPERTIES_METADATA);
+ Assertions.assertTrue(ops.dropSchema(schemaIdent, true));
+ Assertions.assertFalse(store.exists(schemaIdent,
Entity.EntityType.SCHEMA));
+ Assertions.assertFalse(store.exists(filesetIdent,
Entity.EntityType.FILESET));
+ Assertions.assertTrue(fs.exists(path));
+ Assertions.assertThrows(IllegalArgumentException.class, () ->
secretManager.readSecret(urn));
+ }
+ }
+
@ParameterizedTest
@MethodSource("locationArguments")
public void testCreateLoadAndDeleteFilesetWithLocations(
@@ -4158,6 +4618,22 @@ public class TestFilesetCatalogOperations {
}
}
+ private void createFunction(NameIdentifier ident) {
+ new ManagedFunctionOperations(store, idGenerator)
+ .registerFunction(
+ ident,
+ "comment",
+ FunctionType.SCALAR,
+ true,
+ FunctionDefinitions.of(
+ FunctionDefinitions.of(
+ new FunctionParam[0],
+ Types.IntegerType.get(),
+ new FunctionImpl[] {
+ FunctionImpls.ofJava(FunctionImpl.RuntimeType.SPARK,
"com.example.TestUDF")
+ })));
+ }
+
private long generateTestId() {
return idGenerator.nextId();
}
diff --git a/core/src/main/java/org/apache/gravitino/EntityStore.java
b/core/src/main/java/org/apache/gravitino/EntityStore.java
index 62b12ab320..4ed86ff207 100644
--- a/core/src/main/java/org/apache/gravitino/EntityStore.java
+++ b/core/src/main/java/org/apache/gravitino/EntityStore.java
@@ -29,6 +29,7 @@ import java.util.function.Function;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.gravitino.Entity.EntityType;
import org.apache.gravitino.exceptions.NoSuchEntityException;
+import org.apache.gravitino.meta.SchemaEntity;
import org.apache.gravitino.utils.Executable;
public interface EntityStore extends Closeable {
@@ -300,6 +301,26 @@ public interface EntityStore extends Closeable {
}
}
+ /**
+ * Deletes a schema and optionally its functions, then runs an action before
committing.
+ *
+ * <p>Any remaining children, including filesets, prevent deletion. Function
cleanup and the
+ * schema deletion must roll back together if the action or the
remaining-child check fails.
+ *
+ * @param ident the schema identifier
+ * @param cascadeFunctions whether to delete functions in the same metadata
transaction
+ * @param postDeleteAction the action to run against the deleted snapshot
before commit
+ * @return the deleted schema, or empty when it did not exist
+ * @throws IOException if the delete operation fails
+ * @throws UnsupportedOperationException if the store cannot honor this
transaction boundary
+ */
+ default Optional<SchemaEntity> deleteSchemaAndGet(
+ NameIdentifier ident, boolean cascadeFunctions, Consumer<SchemaEntity>
postDeleteAction)
+ throws IOException {
+ throw new UnsupportedOperationException(
+ "This store cannot delete a schema with a pre-commit action");
+ }
+
/**
* 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 096bbdadd8..83b6c2409f 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
@@ -32,6 +32,7 @@ import java.util.Map;
import java.util.Optional;
import java.util.function.Consumer;
import java.util.function.Function;
+import java.util.function.Supplier;
import java.util.stream.Collectors;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.gravitino.Config;
@@ -421,37 +422,22 @@ public class JDBCBackend implements RelationalBackend,
SupportsOrphanedRelationC
if (entityType != Entity.EntityType.FILESET) {
return RelationalBackend.super.deleteAndGet(ident, entityType, clazz,
postDeleteAction);
}
+ return deleteWithCallback(
+ ident,
+ entityType,
+ () ->
clazz.cast(FilesetMetaService.getInstance().deleteFilesetAndGet(ident)),
+ 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();
- }
- EntityChangeLogWriter.append(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 Optional<SchemaEntity> deleteSchemaAndGet(
+ NameIdentifier ident, boolean cascadeFunctions, Consumer<SchemaEntity>
postDeleteAction)
+ throws IOException {
+ return deleteWithCallback(
+ ident,
+ Entity.EntityType.SCHEMA,
+ () -> SchemaMetaService.getInstance().deleteSchemaAndGet(ident, false,
cascadeFunctions),
+ postDeleteAction);
}
@Override
@@ -873,6 +859,42 @@ public class JDBCBackend implements RelationalBackend,
SupportsOrphanedRelationC
}
}
+ private <E extends Entity & HasIdentifier> Optional<E> deleteWithCallback(
+ NameIdentifier ident,
+ Entity.EntityType entityType,
+ Supplier<E> deleteAction,
+ Consumer<E> postDeleteAction) {
+ boolean transactionOwner = !SessionUtils.isInTransaction();
+ if (transactionOwner) {
+ SessionUtils.beginTransaction();
+ }
+ boolean committed = false;
+ try {
+ E deletedEntity;
+ try {
+ deletedEntity = deleteAction.get();
+ } catch (NoSuchEntityException e) {
+ // Only the delete itself may report the entity 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();
+ }
+ EntityChangeLogWriter.append(ident, entityType, OperateType.DROP);
+ // 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();
+ }
+ }
+ }
+
private static NameIdentifier[] toNameIdentifiers(RelationEdgeTarget[]
relationTargets) {
if (relationTargets == null) {
return null;
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 664dd17ec9..fa64b409dc 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
@@ -34,6 +34,7 @@ import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.Namespace;
import org.apache.gravitino.SupportsRelationOperations;
import org.apache.gravitino.exceptions.NoSuchEntityException;
+import org.apache.gravitino.meta.SchemaEntity;
/** Interface defining the operations for a Relation Backend. */
public interface RelationalBackend extends Closeable,
SupportsRelationOperations {
@@ -195,6 +196,26 @@ public interface RelationalBackend extends Closeable,
SupportsRelationOperations
}
}
+ /**
+ * Deletes a schema and optionally its functions, then runs an action before
commit.
+ *
+ * <p>Remaining children, including filesets, must prevent deletion.
Function cleanup, schema
+ * deletion, and the action share one metadata transaction.
+ *
+ * @param ident the schema identifier
+ * @param cascadeFunctions whether to delete functions in the same metadata
transaction
+ * @param postDeleteAction the action to run against the deleted snapshot
before commit
+ * @return the deleted schema, or empty when it did not exist
+ * @throws IOException if the store operation fails
+ * @throws UnsupportedOperationException if the backend cannot honor this
transaction boundary
+ */
+ default Optional<SchemaEntity> deleteSchemaAndGet(
+ NameIdentifier ident, boolean cascadeFunctions, Consumer<SchemaEntity>
postDeleteAction)
+ throws IOException {
+ throw new UnsupportedOperationException(
+ "This backend cannot delete a schema with a pre-commit action");
+ }
+
/**
* 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 876e0c6c84..2a596f8f54 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
@@ -58,6 +58,7 @@ import org.apache.gravitino.cache.EntityCacheKey;
import org.apache.gravitino.cache.NoOpsCache;
import org.apache.gravitino.exceptions.NoSuchEntityException;
import org.apache.gravitino.exceptions.OptimisticLockException;
+import org.apache.gravitino.meta.SchemaEntity;
import org.apache.gravitino.metrics.MetricsSystem;
import org.apache.gravitino.metrics.source.EntityChangeLogMetricsSource;
import org.apache.gravitino.storage.relational.service.EntityIdService;
@@ -334,6 +335,19 @@ public class RelationalEntityStore
}
}
+ @Override
+ public Optional<SchemaEntity> deleteSchemaAndGet(
+ NameIdentifier ident, boolean cascadeFunctions, Consumer<SchemaEntity>
postDeleteAction)
+ throws IOException {
+ try {
+ return backend.deleteSchemaAndGet(ident, cascadeFunctions,
postDeleteAction);
+ } catch (RuntimeException re) {
+ throw ExceptionUtils.lockWaitTimeoutAsConflict(re,
Entity.EntityType.SCHEMA, ident);
+ } finally {
+ invalidateCache(ident, Entity.EntityType.SCHEMA);
+ }
+ }
+
@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/service/SchemaMetaService.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/SchemaMetaService.java
index 7e6f3990af..951f3c799b 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/service/SchemaMetaService.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/SchemaMetaService.java
@@ -43,6 +43,7 @@ import
org.apache.gravitino.exceptions.NonEmptyEntityException;
import org.apache.gravitino.meta.SchemaEntity;
import org.apache.gravitino.metrics.Monitored;
import org.apache.gravitino.storage.IdGenerator;
+import org.apache.gravitino.storage.relational.EntityChangeLogWriter;
import org.apache.gravitino.storage.relational.helper.SchemaIds;
import org.apache.gravitino.storage.relational.mapper.CatalogMetaMapper;
import org.apache.gravitino.storage.relational.mapper.FilesetMetaMapper;
@@ -66,7 +67,9 @@ import
org.apache.gravitino.storage.relational.mapper.TopicMetaMapper;
import org.apache.gravitino.storage.relational.mapper.ViewMetaMapper;
import org.apache.gravitino.storage.relational.mapper.ViewVersionInfoMapper;
import org.apache.gravitino.storage.relational.po.CatalogPO;
+import org.apache.gravitino.storage.relational.po.FunctionPO;
import org.apache.gravitino.storage.relational.po.SchemaPO;
+import org.apache.gravitino.storage.relational.po.cache.OperateType;
import org.apache.gravitino.storage.relational.utils.ExceptionUtils;
import org.apache.gravitino.storage.relational.utils.POConverters;
import org.apache.gravitino.storage.relational.utils.SessionUtils;
@@ -272,9 +275,25 @@ public class SchemaMetaService {
metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
baseMetricName = "deleteSchema")
public boolean deleteSchema(NameIdentifier identifier, boolean cascade) {
+ deleteSchemaAndGet(identifier, cascade, false);
+ return true;
+ }
+
+ /**
+ * Deletes a schema and returns the exact entity snapshot protected by the
delete CAS.
+ *
+ * @param identifier the schema identifier
+ * @param cascade whether to delete child entities
+ * @param cascadeFunctions whether a non-cascading delete should remove
functions before checking
+ * for remaining children
+ * @return the schema snapshot that was deleted
+ */
+ public SchemaEntity deleteSchemaAndGet(
+ NameIdentifier identifier, boolean cascade, boolean cascadeFunctions) {
NameIdentifierUtil.checkSchema(identifier);
SchemaPO schemaPO = getSchemaPOByIdentifier(identifier);
+ SchemaEntity deletedSchema = POConverters.fromSchemaPO(schemaPO,
identifier.namespace());
Long schemaId = schemaPO.getSchemaId();
if (cascade) {
@@ -380,6 +399,11 @@ public class SchemaMetaService {
// to appear in between. A non-empty result throws, which rolls
the delete back.
lockCatalogForSchemaDelete(identifier, schemaPO);
deleteSchemaWithVersion(identifier, schemaPO);
+ if (cascadeFunctions) {
+ // The schema row lock excludes new children. Delete functions
in this transaction
+ // so a remaining fileset or a later cleanup failure restores
them with the schema.
+ deleteSchemaFunctions(identifier, schemaId);
+ }
checkSchemaIsEmpty(identifier, schemaPO);
},
() ->
@@ -405,7 +429,7 @@ public class SchemaMetaService {
StatisticMetaMapper.class,
mapper -> mapper.softDeleteStatisticsByEntityId(schemaId)));
}
- return true;
+ return deletedSchema;
}
/**
@@ -615,6 +639,30 @@ public class SchemaMetaService {
mapper -> mapper.softDeleteSchemaMetasWithVersion(children)));
}
+ private void deleteSchemaFunctions(NameIdentifier schemaIdentifier, Long
schemaId) {
+ try {
+ // Resolve by ID because the schema row has already been soft-deleted in
this transaction.
+ List<FunctionPO> functions =
+ SessionUtils.getWithoutCommit(
+ FunctionMetaMapper.class, mapper ->
mapper.listFunctionPOsBySchemaId(schemaId));
+ Namespace namespace =
+ NamespaceUtil.ofFunction(
+ schemaIdentifier.namespace().level(0),
+ schemaIdentifier.namespace().level(1),
+ schemaIdentifier.name());
+ for (FunctionPO function : functions) {
+ NameIdentifier functionIdentifier = NameIdentifier.of(namespace,
function.functionName());
+
FunctionMetaService.getInstance().deleteFunctionWithVersion(functionIdentifier,
function);
+ EntityChangeLogWriter.append(
+ functionIdentifier, Entity.EntityType.FUNCTION, OperateType.DROP);
+ }
+ } catch (NoSuchEntityException e) {
+ // A missing function must not be reported as a missing schema. Other
failures, including
+ // optimistic conflicts, retain their original exception type.
+ throw new RuntimeException("Failed to delete functions under schema " +
schemaIdentifier, e);
+ }
+ }
+
/** Checks that no active schema or metadata object is left below the
schema. */
private void checkSchemaIsEmpty(NameIdentifier identifier, SchemaPO
schemaPO) {
boolean hasDescendantSchemas =
!listDescendantSchemaPOs(schemaPO).isEmpty();
diff --git
a/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java
b/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java
index d359d43dce..e86579c433 100644
--- a/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java
+++ b/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java
@@ -54,6 +54,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
+import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.commons.lang3.reflect.FieldUtils;
import org.apache.gravitino.Catalog;
import org.apache.gravitino.CatalogChange;
@@ -129,6 +130,7 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.Mockito;
import org.mockito.stubbing.Answer;
@@ -1721,9 +1723,11 @@ public class TestCatalogManager {
Mockito.verify(catalogWrapper, Mockito.never()).doWithSchemaOps(any());
}
- @Test
- public void testForceDropManagedCatalogPropagatesSchemaConflict() throws
Exception {
- NameIdentifier ident = NameIdentifier.of("metalake",
"test_force_drop_occ");
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testForceDropManagedCatalogPreservesMetadataOnSchemaFailure(boolean
conflict)
+ throws Exception {
+ NameIdentifier ident = NameIdentifier.of("metalake",
"test_force_drop_failure_" + conflict);
Map<String, String> props =
ImmutableMap.of(
"provider",
@@ -1740,7 +1744,7 @@ public class TestCatalogManager {
SchemaEntity.builder()
.withId(RandomIdGenerator.INSTANCE.nextId())
.withName("schema_occ")
- .withNamespace(Namespace.of("metalake", "test_force_drop_occ"))
+ .withNamespace(Namespace.of("metalake", ident.name()))
.withAuditInfo(
AuditInfo.builder()
.withCreator(PrincipalUtils.getCurrentPrincipal().getName())
@@ -1755,14 +1759,28 @@ public class TestCatalogManager {
// A managed-storage catalog drops each schema through its schema
operations.
Mockito.doReturn(CapabilityResult.SUPPORTED).when(capability).managedStorage(any());
Mockito.doReturn(catalog).when(catalogWrapper).catalog();
- OptimisticLockException conflict = new OptimisticLockException("lock wait
timeout");
- Mockito.doThrow(conflict).when(catalogWrapper).doWithSchemaOps(any());
+ RuntimeException failure =
+ conflict
+ ? new OptimisticLockException("lock wait timeout")
+ : new RuntimeException("Failed to delete fileset storage", new
IOException("denied"));
+ Mockito.doThrow(failure).when(catalogWrapper).doWithSchemaOps(any());
catalogManager.getCatalogCache().put(ident, catalogWrapper);
- Assertions.assertSame(
- conflict,
+ RuntimeException thrown =
Assertions.assertThrows(
- OptimisticLockException.class, () ->
catalogManager.dropCatalog(ident, true)));
+ RuntimeException.class, () -> catalogManager.dropCatalog(ident,
true));
+ if (conflict) {
+ Assertions.assertSame(failure, thrown);
+ } else {
+ String schemaError =
+ "Failed to drop schema " + NameIdentifier.of("metalake",
ident.name(), "schema_occ");
+ Assertions.assertTrue(thrown.getMessage().contains(schemaError));
+
Assertions.assertTrue(ExceptionUtils.getThrowableList(thrown).contains(failure));
+ }
+ Assertions.assertTrue(entityStore.exists(ident, EntityType.CATALOG));
+ Assertions.assertTrue(
+ entityStore.exists(
+ NameIdentifier.of("metalake", ident.name(), "schema_occ"),
EntityType.SCHEMA));
}
@Test
diff --git
a/core/src/test/java/org/apache/gravitino/metalake/TestMetalakeManager.java
b/core/src/test/java/org/apache/gravitino/metalake/TestMetalakeManager.java
index d27b096c55..37ff307eb4 100644
--- a/core/src/test/java/org/apache/gravitino/metalake/TestMetalakeManager.java
+++ b/core/src/test/java/org/apache/gravitino/metalake/TestMetalakeManager.java
@@ -388,6 +388,9 @@ public class TestMetalakeManager {
Assertions.assertFalse(
MetalakeManager.metalakeInUse(entityStore, ident),
"failed force drop must keep the user-disabled metalake disabled");
+ Assertions.assertTrue(entityStore.exists(ident, EntityType.METALAKE));
+ Assertions.assertTrue(
+ entityStore.exists(NameIdentifier.of(ident.name(), "catalog1"),
EntityType.CATALOG));
} finally {
FieldUtils.writeField(
GravitinoEnv.getInstance(), "catalogManager",
originalEnvCatalogManager, true);
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestSchemaMetaService.java
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestSchemaMetaService.java
index cda36c09f7..3f342ecf3e 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestSchemaMetaService.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestSchemaMetaService.java
@@ -72,9 +72,13 @@ import org.apache.gravitino.storage.RandomIdGenerator;
import org.apache.gravitino.storage.relational.RelationalBackend;
import org.apache.gravitino.storage.relational.TestJDBCBackend;
import org.apache.gravitino.storage.relational.mapper.CatalogMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.EntityChangeLogMapper;
import org.apache.gravitino.storage.relational.mapper.SchemaMetaMapper;
import org.apache.gravitino.storage.relational.po.CatalogPO;
+import org.apache.gravitino.storage.relational.po.FunctionPO;
import org.apache.gravitino.storage.relational.po.SchemaPO;
+import org.apache.gravitino.storage.relational.po.cache.EntityChangeRecord;
+import org.apache.gravitino.storage.relational.po.cache.OperateType;
import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
import org.apache.gravitino.storage.relational.utils.POConverters;
import org.apache.gravitino.storage.relational.utils.SessionUtils;
@@ -83,12 +87,175 @@ import org.apache.gravitino.utils.NamespaceUtil;
import org.apache.ibatis.session.SqlSession;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.TestTemplate;
+import org.mockito.MockedStatic;
import org.mockito.Mockito;
public class TestSchemaMetaService extends TestJDBCBackend {
private final String metalakeName = "metalake_for_catalog_test";
private final String catalogName = "catalog_for_catalog_test";
+ @TestTemplate
+ void testDeleteSchemaAndGetRollsBackWhenCleanupFails() throws IOException {
+ createAndInsertMakeLake(metalakeName);
+ createAndInsertCatalog(metalakeName, catalogName);
+ SchemaEntity schema =
+ createAndInsertSchema(metalakeName, catalogName,
"schema_cleanup_failure");
+ NameIdentifier ident = schema.nameIdentifier();
+ FunctionEntity function =
+ createFunctionEntity(
+ RandomIdGenerator.INSTANCE.nextId(),
+ NamespaceUtil.ofFunction(metalakeName, catalogName, schema.name()),
+ "function",
+ AUDIT_INFO);
+ backend.insert(function, false);
+ long lastChangeId =
+ SessionUtils.getWithoutCommit(
+ EntityChangeLogMapper.class,
EntityChangeLogMapper::selectMaxChangeId);
+ NoSuchEntityException failure = new NoSuchEntityException("cleanup
dependency is missing");
+
+ assertThrows(
+ NonEmptyEntityException.class,
+ () ->
+ backend.deleteSchemaAndGet(
+ ident,
+ false,
+ ignored -> Assertions.fail("A non-cascading drop must preserve
functions")));
+
+ Assertions.assertSame(
+ failure,
+ assertThrows(
+ NoSuchEntityException.class,
+ () ->
+ backend.deleteSchemaAndGet(
+ ident,
+ true,
+ deleted -> {
+ Assertions.assertEquals(schema.id(), deleted.id());
+ Assertions.assertEquals(schema.properties(),
deleted.properties());
+ throw failure;
+ })));
+ assertTrue(backend.exists(ident, Entity.EntityType.SCHEMA));
+ Assertions.assertEquals(
+ function.comment(),
+ backend
+ .<FunctionEntity>get(function.nameIdentifier(),
Entity.EntityType.FUNCTION)
+ .comment());
+ assertTrue(
+ SessionUtils.getWithoutCommit(
+ EntityChangeLogMapper.class,
+ mapper -> mapper.selectEntityChanges(lastChangeId, 100))
+ .isEmpty());
+ Assertions.assertEquals(
+ schema.properties(),
+ backend.<SchemaEntity>get(ident,
Entity.EntityType.SCHEMA).properties());
+
+ SchemaEntity deleted =
+ backend
+ .deleteSchemaAndGet(
+ ident, true, snapshot -> Assertions.assertEquals(schema.id(),
snapshot.id()))
+ .orElseThrow();
+ Assertions.assertEquals(schema.id(), deleted.id());
+ assertFalse(backend.exists(ident, Entity.EntityType.SCHEMA));
+ assertFalse(backend.exists(function.nameIdentifier(),
Entity.EntityType.FUNCTION));
+ List<EntityChangeRecord> changes =
+ SessionUtils.getWithoutCommit(
+ EntityChangeLogMapper.class, mapper ->
mapper.selectEntityChanges(lastChangeId, 100));
+ Assertions.assertEquals(2, changes.size());
+ Assertions.assertEquals(
+ Set.of(Entity.EntityType.SCHEMA.name(),
Entity.EntityType.FUNCTION.name()),
+
changes.stream().map(EntityChangeRecord::getEntityType).collect(Collectors.toSet()));
+ assertTrue(changes.stream().allMatch(change -> change.getOperateType() ==
OperateType.DROP));
+ assertFalse(
+ backend
+ .deleteSchemaAndGet(
+ ident,
+ true,
+ ignored -> Assertions.fail("Cleanup must not run for a missing
schema"))
+ .isPresent());
+ }
+
+ @TestTemplate
+ void testDeleteSchemaAndGetRejectsRemainingFilesetsBeforeCleanup() throws
IOException {
+ createAndInsertMakeLake(metalakeName);
+ createAndInsertCatalog(metalakeName, catalogName);
+ SchemaEntity schema = createAndInsertSchema(metalakeName, catalogName,
"schema_with_child");
+ FilesetEntity fileset =
+ createFilesetEntity(
+ RandomIdGenerator.INSTANCE.nextId(),
+ Namespace.of(metalakeName, catalogName, schema.name()),
+ "fileset",
+ AUDIT_INFO);
+ backend.insert(fileset, false);
+ FunctionEntity function =
+ createFunctionEntity(
+ RandomIdGenerator.INSTANCE.nextId(),
+ NamespaceUtil.ofFunction(metalakeName, catalogName, schema.name()),
+ "function",
+ AUDIT_INFO);
+ backend.insert(function, false);
+
+ assertThrows(
+ NonEmptyEntityException.class,
+ () ->
+ backend.deleteSchemaAndGet(
+ schema.nameIdentifier(),
+ true,
+ ignored -> Assertions.fail("Cleanup must not run for a
non-empty schema")));
+ assertTrue(backend.exists(schema.nameIdentifier(),
Entity.EntityType.SCHEMA));
+ assertTrue(backend.exists(fileset.nameIdentifier(),
Entity.EntityType.FILESET));
+ assertTrue(backend.exists(function.nameIdentifier(),
Entity.EntityType.FUNCTION));
+ }
+
+ @TestTemplate
+ void testDeleteSchemaAndGetPropagatesFunctionCleanupFailure() throws
IOException {
+ createAndInsertMakeLake(metalakeName);
+ createAndInsertCatalog(metalakeName, catalogName);
+ SchemaEntity schema =
+ createAndInsertSchema(metalakeName, catalogName,
"function_cleanup_failure");
+ FunctionEntity function =
+ createFunctionEntity(
+ RandomIdGenerator.INSTANCE.nextId(),
+ NamespaceUtil.ofFunction(metalakeName, catalogName, schema.name()),
+ "function",
+ AUDIT_INFO);
+ backend.insert(function, false);
+ FunctionMetaService functionService = FunctionMetaService.getInstance();
+
+ for (RuntimeException failure :
+ List.of(
+ new NoSuchEntityException("function disappeared"),
+ new OptimisticLockException("function changed"))) {
+ FunctionMetaService failingService = Mockito.spy(functionService);
+ Mockito.doThrow(failure)
+ .when(failingService)
+ .deleteFunctionWithVersion(
+ Mockito.eq(function.nameIdentifier()),
Mockito.any(FunctionPO.class));
+ try (MockedStatic<FunctionMetaService> mocked =
+ Mockito.mockStatic(FunctionMetaService.class)) {
+
mocked.when(FunctionMetaService::getInstance).thenReturn(failingService);
+ RuntimeException actual =
+ assertThrows(
+ RuntimeException.class,
+ () ->
+ backend.deleteSchemaAndGet(
+ schema.nameIdentifier(),
+ true,
+ ignored ->
+ Assertions.fail("Cleanup must not run after a
function failure")));
+ if (failure instanceof NoSuchEntityException) {
+ Assertions.assertEquals(
+ "Failed to delete functions under schema " +
schema.nameIdentifier(),
+ actual.getMessage());
+ Assertions.assertSame(failure, actual.getCause());
+ } else {
+ Assertions.assertSame(failure, actual);
+ }
+ }
+ assertTrue(backend.exists(schema.nameIdentifier(),
Entity.EntityType.SCHEMA));
+ assertTrue(backend.exists(function.nameIdentifier(),
Entity.EntityType.FUNCTION));
+ }
+ }
+
@TestTemplate
public void testInsertAlreadyExistsException() throws IOException {
createAndInsertMakeLake(metalakeName);
