This is an automated email from the ASF dual-hosted git repository.
jerryshao pushed a commit to branch branch-1.3
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/branch-1.3 by this push:
new bf4136ce37 [Cherry-pick to branch-1.3] [#14119] fix(fileset): Preserve
metadata on cascade cleanup failure (#14120) (#14203)
bf4136ce37 is described below
commit bf4136ce37d1d5f19c424179de9650664e060dc1
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Mon Oct 26 17:08:59 2026 +0800
[Cherry-pick to branch-1.3] [#14119] fix(fileset): Preserve metadata on
cascade cleanup failure (#14120) (#14203)
### What changes were proposed in this pull request?
Backports #14120 ([original
commit](https://github.com/apache/gravitino/commit/c816482b8d24d4a77aaa32c87434398f2d030302))
to `branch-1.3` and resolves the cherry-pick conflicts.
Managed filesets are deleted before their schema, stopping at the first
storage failure. Schema and function metadata remain in one transaction
until empty schema-directory cleanup succeeds. The implementation and
tests use the existing 1.3 deletion, transaction, and change-log APIs.
### Why are the changes needed?
Catalog and metalake force-drop can bypass the normal schema dispatcher.
Previously, metadata was deleted before managed storage cleanup, leaving
failed cleanup without metadata for retry.
Fix: #14119
### Does this PR introduce _any_ user-facing change?
Cleanup failures retain the failed fileset and its parents, or
schema/function metadata when schema-directory cleanup fails. Earlier
successful fileset deletions remain committed, and retries accept
already-removed directories. 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`: 252 tests passed.
- `:core:test`: 82 focused tests passed across schema/fileset metadata
services, catalog/metalake managers, schema/fileset dispatchers, and
owner events; relational tests used H2.
- Tests used `-PskipITs -PskipDockerTests=true`.
- `spotlessApply`, `git diff --check`, and conflict-marker scans passed.
- Docker/MySQL/PostgreSQL integration tests were not run.
---------
Co-authored-by: mchades <[email protected]>
Co-authored-by: Jerry Shao <[email protected]>
---
.../catalog/fileset/FilesetCatalogOperations.java | 238 +++++-------
.../fileset/TestFilesetCatalogOperations.java | 417 +++++++++++++++++++++
.../java/org/apache/gravitino/EntityStore.java | 23 ++
.../gravitino/storage/relational/JDBCBackend.java | 28 ++
.../storage/relational/RelationalBackend.java | 23 ++
.../storage/relational/RelationalEntityStore.java | 13 +
.../relational/service/SchemaMetaService.java | 37 ++
.../gravitino/catalog/TestCatalogManager.java | 55 +++
.../listener/api/event/TestOwnerEvent.java | 9 +
.../gravitino/metalake/TestMetalakeManager.java | 47 +++
.../relational/service/TestSchemaMetaService.java | 183 +++++++++
11 files changed, 925 insertions(+), 148 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 62db7197de..f16c3329a9 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;
@@ -106,6 +105,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;
@@ -701,50 +701,10 @@ 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);
- }
- } 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();
- }
+ FilesetEntity fileset = store.get(ident, Entity.EntityType.FILESET,
FilesetEntity.class);
+ if (!disableFSOps && fileset.filesetType() == Fileset.Type.MANAGED) {
+ deleteManagedFilesetStorage(ident, fileset);
}
-
return store.delete(ident, Entity.EntityType.FILESET);
} catch (NoSuchEntityException ne) {
LOG.warn("Fileset {} does not exist", ident);
@@ -866,118 +826,45 @@ 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) {
- 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;
+ // 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) {
+ dropFileset(fileset.nameIdentifier());
+ }
}
- // 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);
+ // 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);
+ }
}
- }
- }
- } 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();
- }
+ })
+ .isPresent();
+ if (dropped) {
+ LOG.info("Deleted schema {}", ident);
}
+ return dropped;
- LOG.info("Deleted schema {}", ident);
- return true;
-
+ } 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);
}
@@ -1052,6 +939,61 @@ public class FilesetCatalogOperations extends
ManagedSchemaOperations
ClassLoaderResourceCleanerUtils.closeClassLoaderResource(this.getClass().getClassLoader());
}
+ /** Stops at the first failed location, leaving fileset metadata available
for retry. */
+ 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));
+ }
+ }
+ }
+
+ 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 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 da358bcc46..51e53dd98d 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
@@ -58,9 +58,11 @@ import java.io.IOException;
import java.net.ConnectException;
import java.nio.file.Files;
import java.nio.file.Paths;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
@@ -74,6 +76,7 @@ import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.reflect.FieldUtils;
import org.apache.gravitino.Config;
import org.apache.gravitino.Configs;
+import org.apache.gravitino.Entity;
import org.apache.gravitino.EntityAlreadyExistsException;
import org.apache.gravitino.EntityFieldLimits;
import org.apache.gravitino.EntityStore;
@@ -88,6 +91,7 @@ import org.apache.gravitino.UserPrincipal;
import org.apache.gravitino.audit.CallerContext;
import org.apache.gravitino.audit.FilesetAuditConstants;
import org.apache.gravitino.audit.FilesetDataOperation;
+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;
@@ -109,6 +113,13 @@ 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.FilesetEntity;
+import org.apache.gravitino.rel.types.Types;
import org.apache.gravitino.storage.IdGenerator;
import org.apache.gravitino.storage.RandomIdGenerator;
import org.apache.gravitino.storage.relational.RelationalEntityStore;
@@ -120,6 +131,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;
@@ -133,6 +145,7 @@ import org.junit.jupiter.api.Timeout;
import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.CsvSource;
import org.junit.jupiter.params.provider.EnumSource;
import org.junit.jupiter.params.provider.MethodSource;
import org.junit.jupiter.params.provider.ValueSource;
@@ -792,6 +805,394 @@ 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();
+
+ 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))) {
+ 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-prop",
"schema-value")));
+ 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));
+ Mockito.verify(ops)
+ .getFileSystemWithCache(
+ Mockito.eq(schemaPath),
+ Mockito.argThat(fsConf ->
"schema-value".equals(fsConf.get("schema-prop"))));
+
+ 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));
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testDropSchemaWhenDirectoryDisappearsDuringDelete(boolean
throwException)
+ throws IOException {
+ String schemaName = "schema_directory_disappeared_" + generateTestId();
+ String catalogPath = TEST_ROOT_PATH + "/" + schemaName;
+ createSchema(generateTestId(), 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))) {
+ 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(generateTestId(), 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)) {
+ 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(generateTestId(), 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<>();
+ for (String name : List.of("first", "failed", "last")) {
+ createFileset(name, schemaName, "comment", Fileset.Type.MANAGED,
catalogPath, null);
+ 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))) {
+ 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));
+ }
+
+ // 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));
+ }
+ }
+ }
+
+ @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(generateTestId(), schemaName, "comment", catalogPath, null);
+ NameIdentifier schemaIdent = NameIdentifier.of("m1", "c1", schemaName);
+ NameIdentifier filesetIdent = NameIdentifier.of("m1", "c1", schemaName,
"fileset");
+ 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));
+ 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))) {
+ 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());
+
+ // 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));
+ }
+ }
+ }
+
+ @Test
+ void testDropSchemaRejectsFilesetCreatedAfterListing() throws IOException {
+ String schemaName = "schema_cascade_new_fileset_" + generateTestId();
+ String catalogPath = TEST_ROOT_PATH + "/" + schemaName;
+ createSchema(generateTestId(), 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();
+ Map<String, String> remainingProperties =
+ StringIdentifier.newPropertiesWithId(
+ StringIdentifier.fromId(remainingId), Collections.emptyMap());
+
+ EntityStore interleavedStore = Mockito.spy(store);
+ try (FilesetCatalogOperations ops = new
FilesetCatalogOperations(interleavedStore)) {
+ 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.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));
+ }
+ }
+
+ @ParameterizedTest
+ @CsvSource({"EXTERNAL, false", "MANAGED, true"})
+ void testDropSchemaPreservesExternalOrDisabledStorage(Fileset.Type type,
boolean disableFSOps)
+ throws IOException {
+ String schemaName = "schema_cascade_metadata_only_" + generateTestId();
+ String catalogPath = TEST_ROOT_PATH + "/" + schemaName;
+ createSchema(generateTestId(), schemaName, "comment", catalogPath, null);
+ NameIdentifier schemaIdent = NameIdentifier.of("m1", "c1", schemaName);
+ NameIdentifier filesetIdent = NameIdentifier.of("m1", "c1", schemaName,
"fileset");
+ Fileset fileset =
+ createFileset(
+ filesetIdent.name(),
+ schemaName,
+ "comment",
+ type,
+ catalogPath,
+ catalogPath + "/fileset");
+ Path path = new Path(fileset.storageLocation());
+ FileSystem fs = path.getFileSystem(new Configuration());
+ Assertions.assertTrue(fs.exists(path));
+
+ try (FilesetCatalogOperations ops = new FilesetCatalogOperations(store)) {
+ 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));
+ }
+ }
+
@ParameterizedTest
@MethodSource("locationArguments")
public void testCreateLoadAndDeleteFilesetWithLocations(
@@ -3759,6 +4160,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 d2c19c65c6..8c63246f71 100644
--- a/core/src/main/java/org/apache/gravitino/EntityStore.java
+++ b/core/src/main/java/org/apache/gravitino/EntityStore.java
@@ -23,10 +23,13 @@ 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;
import org.apache.gravitino.exceptions.NoSuchEntityException;
+import org.apache.gravitino.meta.SchemaEntity;
import org.apache.gravitino.utils.Executable;
public interface EntityStore extends Closeable {
@@ -221,6 +224,26 @@ public interface EntityStore extends Closeable {
*/
boolean delete(NameIdentifier ident, EntityType entityType, boolean cascade)
throws IOException;
+ /**
+ * Deletes a schema and optionally its functions, then runs an action before
committing.
+ *
+ * <p>Remaining 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 65182cf57e..96d8856ae7 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
@@ -28,6 +28,8 @@ import com.google.common.collect.Lists;
import java.io.IOException;
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;
@@ -84,6 +86,7 @@ import
org.apache.gravitino.storage.relational.service.TopicMetaService;
import org.apache.gravitino.storage.relational.service.UserMetaService;
import org.apache.gravitino.storage.relational.service.ViewMetaService;
import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
+import org.apache.gravitino.storage.relational.utils.SessionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -435,6 +438,31 @@ public class JDBCBackend implements RelationalBackend {
}
}
+ @Override
+ public Optional<SchemaEntity> deleteSchemaAndGet(
+ NameIdentifier ident, boolean cascadeFunctions, Consumer<SchemaEntity>
postDeleteAction)
+ throws IOException {
+ SessionUtils.beginTransaction();
+ boolean committed = false;
+ try {
+ SchemaEntity deletedSchema;
+ try {
+ deletedSchema =
SchemaMetaService.getInstance().deleteSchemaAndGet(ident, cascadeFunctions);
+ } catch (NoSuchEntityException e) {
+ return Optional.empty();
+ }
+ // Keep schema and function metadata recoverable until directory cleanup
succeeds.
+ postDeleteAction.accept(deletedSchema);
+ SessionUtils.commitTransaction();
+ committed = true;
+ return Optional.of(deletedSchema);
+ } finally {
+ if (!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 61eb9c435b..74e2b1fedc 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,6 +21,8 @@ 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;
@@ -31,6 +33,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 {
@@ -137,6 +140,26 @@ public interface RelationalBackend extends Closeable,
SupportsRelationOperations
boolean delete(NameIdentifier ident, Entity.EntityType entityType, boolean
cascade)
throws IOException;
+ /**
+ * Deletes a schema and optionally its functions, then runs an action before
commit.
+ *
+ * <p>Remaining 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 c710b778b0..40cf1b166e 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/RelationalEntityStore.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/RelationalEntityStore.java
@@ -29,6 +29,7 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
import java.util.function.Function;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.gravitino.Config;
@@ -51,6 +52,7 @@ import org.apache.gravitino.cache.NoOpsCache;
import org.apache.gravitino.exceptions.NoSuchEntityException;
import org.apache.gravitino.meta.GroupEntity;
import org.apache.gravitino.meta.RoleEntity;
+import org.apache.gravitino.meta.SchemaEntity;
import org.apache.gravitino.meta.UserEntity;
import org.apache.gravitino.storage.relational.service.EntityIdService;
import org.apache.gravitino.utils.Executable;
@@ -215,6 +217,17 @@ public class RelationalEntityStore
}
}
+ @Override
+ public Optional<SchemaEntity> deleteSchemaAndGet(
+ NameIdentifier ident, boolean cascadeFunctions, Consumer<SchemaEntity>
postDeleteAction)
+ throws IOException {
+ try {
+ return backend.deleteSchemaAndGet(ident, cascadeFunctions,
postDeleteAction);
+ } finally {
+ cache.invalidate(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 39d820b42d..2afa101d85 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
@@ -41,6 +41,7 @@ import org.apache.gravitino.Namespace;
import org.apache.gravitino.exceptions.NoSuchEntityException;
import org.apache.gravitino.exceptions.NonEmptyEntityException;
import org.apache.gravitino.meta.FilesetEntity;
+import org.apache.gravitino.meta.FunctionEntity;
import org.apache.gravitino.meta.ModelEntity;
import org.apache.gravitino.meta.NamespacedEntityId;
import org.apache.gravitino.meta.SchemaEntity;
@@ -264,6 +265,42 @@ public class SchemaMetaService {
}
}
+ /**
+ * Deletes a schema and optionally its functions within the caller's
metadata transaction.
+ *
+ * <p>The caller must keep the transaction open until any external cleanup
succeeds. Remaining
+ * filesets reject the non-cascading schema deletion and roll back function
cleanup with it.
+ *
+ * @param identifier the schema identifier
+ * @param cascadeFunctions whether to remove functions before the
non-cascading schema delete
+ * @return the schema snapshot used for external cleanup
+ */
+ public SchemaEntity deleteSchemaAndGet(NameIdentifier identifier, boolean
cascadeFunctions) {
+ SchemaEntity schema = getSchemaByIdentifier(identifier);
+ try {
+ List<FunctionEntity> functions =
+ FunctionMetaService.getInstance()
+ .listFunctionsByNamespace(
+ NamespaceUtil.ofFunction(
+ identifier.namespace().level(0),
+ identifier.namespace().level(1),
+ identifier.name()));
+ if (!functions.isEmpty() && !cascadeFunctions) {
+ throw new NonEmptyEntityException(
+ "Entity %s has sub-entities, you should remove sub-entities
first", identifier);
+ }
+ for (FunctionEntity function : functions) {
+ if
(!FunctionMetaService.getInstance().deleteFunction(function.nameIdentifier())) {
+ throw new NoSuchEntityException("Function %s does not exist",
function.nameIdentifier());
+ }
+ }
+ } catch (NoSuchEntityException e) {
+ throw new RuntimeException("Failed to delete functions under schema " +
identifier, e);
+ }
+ deleteSchema(identifier, false);
+ return schema;
+ }
+
@Monitored(
metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
baseMetricName = "deleteSchema")
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 d64aff46de..05ca5228b3 100644
--- a/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java
+++ b/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java
@@ -49,6 +49,7 @@ import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
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;
@@ -1343,6 +1344,60 @@ public class TestCatalogManager {
Mockito.verify(catalogWrapper, Mockito.never()).doWithSchemaOps(any());
}
+ @Test
+ void testForceDropManagedCatalogPreservesMetadataOnSchemaFailure() throws
Exception {
+ NameIdentifier ident = NameIdentifier.of("metalake",
"test_force_drop_failure");
+ Map<String, String> props =
+ ImmutableMap.of(
+ "provider",
+ "test",
+ PROPERTY_KEY1,
+ "value1",
+ PROPERTY_KEY2,
+ "value2",
+ PROPERTY_KEY5_PREFIX + "1",
+ "value3");
+ catalogManager.createCatalog(ident, Catalog.Type.RELATIONAL, provider,
"comment", props);
+ BaseCatalog<?> catalog =
catalogManager.loadCatalogAndWrap(ident).catalog();
+ entityStore.put(
+ SchemaEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName("schema_cleanup_failure")
+ .withNamespace(Namespace.of("metalake", ident.name()))
+ .withAuditInfo(
+ AuditInfo.builder()
+
.withCreator(PrincipalUtils.getCurrentPrincipal().getName())
+ .withCreateTime(Instant.now())
+ .build())
+ .build());
+ CatalogManager.CatalogWrapper catalogWrapper =
+ Mockito.mock(CatalogManager.CatalogWrapper.class);
+ Capability capability = Mockito.mock(Capability.class);
+ Mockito.when(catalogWrapper.tryAcquire()).thenReturn(true);
+ Mockito.doReturn(capability).when(catalogWrapper).capabilities();
+ // 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();
+ RuntimeException failure =
+ new RuntimeException("Failed to delete fileset storage", new
IOException("denied"));
+ Mockito.doThrow(failure).when(catalogWrapper).doWithSchemaOps(any());
+ catalogManager.getCatalogCache().put(ident, catalogWrapper);
+
+ RuntimeException thrown =
+ Assertions.assertThrows(
+ RuntimeException.class, () -> catalogManager.dropCatalog(ident,
true));
+ String schemaError =
+ "Failed to drop schema "
+ + NameIdentifier.of("metalake", ident.name(),
"schema_cleanup_failure");
+ 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_cleanup_failure"),
+ EntityType.SCHEMA));
+ }
+
@Test
void testDropCatalogInvalidatesCacheAfterStoreDelete() throws Exception {
NameIdentifier ident = NameIdentifier.of("metalake", "cache_drop_test");
diff --git
a/core/src/test/java/org/apache/gravitino/listener/api/event/TestOwnerEvent.java
b/core/src/test/java/org/apache/gravitino/listener/api/event/TestOwnerEvent.java
index 4f2c597b38..fe88844152 100644
---
a/core/src/test/java/org/apache/gravitino/listener/api/event/TestOwnerEvent.java
+++
b/core/src/test/java/org/apache/gravitino/listener/api/event/TestOwnerEvent.java
@@ -24,8 +24,11 @@ import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.doThrow;
import com.google.common.collect.ImmutableList;
+import java.io.IOException;
import java.util.Arrays;
import java.util.List;
+import java.util.Optional;
+import java.util.function.Consumer;
import org.apache.gravitino.EntityStore;
import org.apache.gravitino.MetadataObject;
import org.apache.gravitino.MetadataObjects;
@@ -37,6 +40,7 @@ import org.apache.gravitino.authorization.OwnerManager;
import org.apache.gravitino.listener.DummyEventListener;
import org.apache.gravitino.listener.EventBus;
import org.apache.gravitino.listener.api.info.OwnerInfo;
+import org.apache.gravitino.meta.SchemaEntity;
import org.apache.gravitino.storage.relational.RelationalBackend;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
@@ -53,6 +57,11 @@ public class TestOwnerEvent {
interface TestEntityStore extends EntityStore, RelationalBackend,
SupportsRelationOperations {
// This interface is used to mock the EntityStore for testing purposes.
+ /** {@inheritDoc} */
+ @Override
+ Optional<SchemaEntity> deleteSchemaAndGet(
+ NameIdentifier ident, boolean cascadeFunctions, Consumer<SchemaEntity>
postDeleteAction)
+ throws IOException;
}
@BeforeAll
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 d42786c8ee..0791b20a67 100644
--- a/core/src/test/java/org/apache/gravitino/metalake/TestMetalakeManager.java
+++ b/core/src/test/java/org/apache/gravitino/metalake/TestMetalakeManager.java
@@ -339,6 +339,53 @@ public class TestMetalakeManager {
}
}
+ @Test
+ void testForceDropMetalakePreservesMetadataOnCatalogFailure() throws
Exception {
+ InMemoryEntityStore store = new InMemoryEntityStore();
+ store.initialize(config);
+ CatalogManager catalogManager = Mockito.mock(CatalogManager.class);
+ Mockito.when(catalogManager.dropCatalog(Mockito.any(NameIdentifier.class),
Mockito.eq(true)))
+ .thenThrow(new RuntimeException("catalog drop failed"));
+ // enableMetalake/disableMetalake propagate metalake-in-use to catalogs
through
+ // GravitinoEnv.catalogManager(), so point it at the mock for the duration
of this test.
+ Mockito.doNothing()
+ .when(catalogManager)
+ .setMetalakeInUseStatus(Mockito.any(NameIdentifier.class),
Mockito.anyBoolean());
+ Object previousCatalogManager =
+ FieldUtils.readField(GravitinoEnv.getInstance(), "catalogManager",
true);
+ FieldUtils.writeField(GravitinoEnv.getInstance(), "catalogManager",
catalogManager, true);
+
+ try {
+ MetalakeManager manager = new MetalakeManager(store, new
RandomIdGenerator(), catalogManager);
+
+ NameIdentifier metalakeIdent =
NameIdentifier.of("force_drop_failure_ml");
+ manager.createMetalake(metalakeIdent, "comment", ImmutableMap.of());
+ store.put(
+ CatalogEntity.builder()
+ .withId(99L)
+ .withName("leftover_catalog")
+ .withNamespace(Namespace.of(metalakeIdent.name()))
+ .withType(Catalog.Type.RELATIONAL)
+ .withProvider("test")
+ .withAuditInfo(
+
AuditInfo.builder().withCreator("test").withCreateTime(Instant.now()).build())
+ .build(),
+ true);
+
+ Assertions.assertThrows(
+ RuntimeException.class, () -> manager.dropMetalake(metalakeIdent,
true));
+ Assertions.assertTrue(store.exists(metalakeIdent,
Entity.EntityType.METALAKE));
+ Assertions.assertTrue(
+ store.exists(
+ NameIdentifier.of(metalakeIdent.name(), "leftover_catalog"),
+ Entity.EntityType.CATALOG));
+ } finally {
+ FieldUtils.writeField(
+ GravitinoEnv.getInstance(), "catalogManager",
previousCatalogManager, true);
+ store.close();
+ }
+ }
+
private void testProperties(Map<String, String> expectedProps, Map<String,
String> testProps) {
expectedProps.forEach(
(k, v) -> {
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 a1e43144ec..3b4241a99a 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
@@ -36,6 +36,7 @@ import org.apache.gravitino.Entity;
import org.apache.gravitino.EntityAlreadyExistsException;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.Namespace;
+import org.apache.gravitino.exceptions.NoSuchEntityException;
import org.apache.gravitino.exceptions.NonEmptyEntityException;
import org.apache.gravitino.meta.ColumnEntity;
import org.apache.gravitino.meta.FilesetEntity;
@@ -49,17 +50,199 @@ import org.apache.gravitino.meta.ViewEntity;
import org.apache.gravitino.rel.types.Types;
import org.apache.gravitino.storage.RandomIdGenerator;
import org.apache.gravitino.storage.relational.TestJDBCBackend;
+import org.apache.gravitino.storage.relational.mapper.EntityChangeLogMapper;
+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.SessionUtils;
import org.apache.gravitino.utils.NameIdentifierUtil;
import org.apache.gravitino.utils.NamespaceUtil;
import org.apache.ibatis.session.SqlSession;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.TestTemplate;
+import org.mockito.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 IllegalStateException("function cleanup failed"))) {
+ FunctionMetaService failingService = Mockito.spy(functionService);
+
Mockito.doThrow(failure).when(failingService).deleteFunction(function.nameIdentifier());
+ 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));
+ }
+
+ FunctionMetaService failingService = Mockito.spy(functionService);
+
Mockito.doReturn(false).when(failingService).deleteFunction(function.nameIdentifier());
+ try (MockedStatic<FunctionMetaService> mocked =
Mockito.mockStatic(FunctionMetaService.class)) {
+ mocked.when(FunctionMetaService::getInstance).thenReturn(failingService);
+ RuntimeException failure =
+ assertThrows(
+ RuntimeException.class,
+ () ->
+ backend.deleteSchemaAndGet(
+ schema.nameIdentifier(),
+ true,
+ ignored -> Assertions.fail("Cleanup must not run when a
function remains")));
+ Assertions.assertInstanceOf(NoSuchEntityException.class,
failure.getCause());
+ }
+ assertTrue(backend.exists(schema.nameIdentifier(),
Entity.EntityType.SCHEMA));
+ assertTrue(backend.exists(function.nameIdentifier(),
Entity.EntityType.FUNCTION));
+ }
+
@TestTemplate
public void testInsertAlreadyExistsException() throws IOException {
createAndInsertMakeLake(metalakeName);
