This is an automated email from the ASF dual-hosted git repository.
roryqi 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 2adfe2321b [#11363] feat(iceberg-rest): wire async cleanup into the
REST drop flow (#11364)
2adfe2321b is described below
commit 2adfe2321b388a4c110cfb71ea6f799265425480
Author: roryqi <[email protected]>
AuthorDate: Wed Jun 3 15:08:15 2026 +0800
[#11363] feat(iceberg-rest): wire async cleanup into the REST drop flow
(#11364)
### What changes were proposed in this pull request?
Final PR (3 of 3) of async hard deletion — wires the cleanup engine
(#11298) into the REST path:
- `IcebergTableOperationExecutor.dropTable` branches on the
`X-Gravitino-Async-Purge` header: synchronous purge by default; when
`true` in auxiliary mode it snapshots the table metadata location, drops
the catalog entry, and enqueues an `IcebergCleanupJob` (keyed by catalog
id) instead of deleting files inline.
- Name-reuse tombstone: `createTable`/`registerTable` return `409` while
an active cleanup job occupies the identifier (best-effort — skipped
when no Gravitino catalog entity backs the name).
- Adds
`IcebergCatalogWrapper.loadTableMetadata/fileIOImpl/fileIOProperties`,
`IcebergRequestContext.asyncPurge()`, and `IcebergCleanupManager`
lifecycle wiring in `RESTService`.
- Documents the async semantics, the `X-Gravitino-Async-Purge` header,
and the `async-cleanup.*` config.
### Why are the changes needed?
The async cleanup engine (#11298) is unused until the REST drop path
opts requests into it.
Fix: #11363
### Does this PR introduce _any_ user-facing change?
- New opt-in request header `X-Gravitino-Async-Purge: true` on `DELETE
...?purgeRequested=true`.
- New `gravitino.iceberg-rest.async-cleanup.*` configuration keys
(documented).
- `createTable`/`registerTable` return `409 Conflict` while a cleanup
job is active for the identifier.
### How was this patch tested?
- Unit: `TestIcebergTableOperationExecutorAsyncPurge`,
`TestIcebergPurgeTombstone`, `TestIcebergRequestContext`; full
`:iceberg:iceberg-rest-server` unit suite green.
- Integration: `IcebergRESTAsyncPurgeIT` (Gravitino server + Iceberg
REST aux service, PostgreSQL-backed catalog via the dynamic provider) —
verifies the async drop enqueues a cleanup job, the `409` tombstone
blocks recreate while in flight, and a background worker deletes the
table's files.
---
docs/iceberg-rest-service.md | 20 ++
.../iceberg/common/ops/IcebergCatalogWrapper.java | 32 +++
.../org/apache/gravitino/iceberg/RESTService.java | 31 ++-
.../service/cleanup/IcebergCleanupJobStore.java | 27 +++
.../IcebergCleanupMapperPackageProvider.java | 12 +-
.../service/dispatcher/IcebergCleanupHelper.java | 79 +++++++
.../IcebergNamespaceOperationExecutor.java | 12 +-
.../dispatcher/IcebergTableOperationExecutor.java | 57 ++++-
.../listener/api/event/IcebergRequestContext.java | 21 ++
...elational.mapper.provider.MapperPackageProvider | 19 --
.../integration/test/IcebergRESTAsyncPurgeIT.java | 263 +++++++++++++++++++++
.../service/dispatcher/TestIcebergAsyncPurge.java | 260 ++++++++++++++++++++
.../TestIcebergNamespaceOperationExecutor.java | 3 +-
.../TestIcebergTableOperationExecutor.java | 3 +-
.../iceberg/service/rest/IcebergRestTestUtil.java | 8 +-
.../api/event/TestIcebergRequestContext.java | 89 +++++++
16 files changed, 896 insertions(+), 40 deletions(-)
diff --git a/docs/iceberg-rest-service.md b/docs/iceberg-rest-service.md
index d25832d6cb..837256389b 100644
--- a/docs/iceberg-rest-service.md
+++ b/docs/iceberg-rest-service.md
@@ -91,6 +91,26 @@ Please note that, it only takes affect in `gravitino.conf`,
you don't need to sp
The filter in `customFilters` should be a standard javax servlet filter.
You can also specify filter parameters by setting configuration entries in the
style `gravitino.iceberg-rest.<class name of filter>.param.<param
name>=<value>`.
+### Asynchronous table purge
+
+By default, dropping a table with `purgeRequested=true` is synchronous: the
catalog entry and the table files are removed before the `DELETE` returns.
+
+When the Iceberg REST service runs inside Gravitino (as an auxiliary service),
a client can instead request asynchronous purge by adding the header
`X-Gravitino-Async-Purge: true` to `DELETE ...?purgeRequested=true`. The drop
then returns `204 No Content` once the table is removed from the catalog, and
the files are deleted in the background. The table is gone from `LIST`
immediately, but recreating it (`createTable` / `registerTable` with the same
name) returns `409 Conflict` until the [...]
+
+The header name is case-insensitive (per the HTTP standard), but its value
must be exactly `true`. Any other value, or no header, uses the synchronous
default, so standard Iceberg clients are unaffected. Asynchronous purge is only
available in auxiliary mode; in standalone mode the header is ignored.
+
+The settings below tune the background workers and are optional.
+
+| Configuration item | Description
| Default value | Required | Since Version |
+|---------------------------------------------------------------|------------------------------------------------------------------------------------------------------|---------------|----------|---------------|
+| `gravitino.iceberg-rest.async-cleanup.worker-threads` | Worker pool
size per server. Each worker claims and runs cleanup jobs from the shared
backend table. | `2` | No | 1.3.0 |
+| `gravitino.iceberg-rest.async-cleanup.delete-threads` | Server-wide
file-delete pool size shared by cleanup jobs.
| `4` | No | 1.3.0 |
+| `gravitino.iceberg-rest.async-cleanup.delete-batch-size` | Number of
files per bulk-delete batch.
| `1000` | No | 1.3.0 |
+| `gravitino.iceberg-rest.async-cleanup.poll-interval-secs` | Worker
polling interval in seconds. This also controls retry pacing for pending jobs.
| `5` | No | 1.3.0 |
+| `gravitino.iceberg-rest.async-cleanup.heartbeat-timeout-secs` | Age in
seconds after which a running job with no fresh heartbeat can be reclaimed by
another worker. | `300` | No | 1.3.0 |
+| `gravitino.iceberg-rest.async-cleanup.max-attempts` | Number of
failed attempts before a cleanup job is marked `FAILED`.
| `5` | No | 1.3.0 |
+| `gravitino.iceberg-rest.async-cleanup.retention-hours` | Retention
time for terminal `SUCCEEDED` or `FAILED` cleanup rows before pruning.
| `720` | No | 1.3.0 |
+
### Catalog backend configuration
:::info
diff --git
a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ops/IcebergCatalogWrapper.java
b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ops/IcebergCatalogWrapper.java
index 987fd329f6..529c352cae 100644
---
a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ops/IcebergCatalogWrapper.java
+++
b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ops/IcebergCatalogWrapper.java
@@ -37,6 +37,7 @@ import org.apache.gravitino.utils.ClassUtils;
import org.apache.gravitino.utils.IsolatedClassLoader;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.iceberg.BaseTable;
import org.apache.iceberg.TableMetadata;
import org.apache.iceberg.Transaction;
import org.apache.iceberg.catalog.Catalog;
@@ -44,6 +45,7 @@ import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.SupportsNamespaces;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.catalog.ViewCatalog;
+import org.apache.iceberg.io.ResolvingFileIO;
import org.apache.iceberg.jdbc.JdbcCatalogWithMetadataLocationSupport;
import org.apache.iceberg.rest.CatalogHandlers;
import org.apache.iceberg.rest.RESTCatalog;
@@ -240,6 +242,36 @@ public class IcebergCatalogWrapper implements
AutoCloseable {
CatalogHandlers.purgeTable(getCatalog(), tableIdentifier);
}
+ /**
+ * Loads current {@link TableMetadata}, bypassing the response cache. Used
by the async cleanup
+ * path to snapshot the metadata location before dropping the catalog entry.
+ *
+ * @param tableIdentifier the table
+ * @return its metadata
+ */
+ public TableMetadata loadTableMetadata(TableIdentifier tableIdentifier) {
+ return ((BaseTable)
getCatalog().loadTable(tableIdentifier)).operations().current();
+ }
+
+ /**
+ * Returns the FileIO implementation configured for this catalog.
+ *
+ * @return the {@code io-impl} class, or the Iceberg default when unset
+ */
+ public String fileIOImpl() {
+ String impl = icebergConfig.get(IcebergConfig.IO_IMPL);
+ return StringUtils.isNotBlank(impl) ? impl :
ResolvingFileIO.class.getName();
+ }
+
+ /**
+ * Returns catalog properties used to reconstruct FileIO in a cleanup worker.
+ *
+ * @return catalog properties snapshotted at enqueue time
+ */
+ public Map<String, String> fileIOProperties() {
+ return getIcebergConfig().getIcebergCatalogProperties();
+ }
+
public LoadTableResponse loadTable(TableIdentifier tableIdentifier) {
Optional<TableMetadata> tableMetadataOptional =
getMetadataCache().getTableMetadata(tableIdentifier);
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/RESTService.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/RESTService.java
index 17b80ea8bc..ea2b54fa6f 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/RESTService.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/RESTService.java
@@ -21,6 +21,7 @@ package org.apache.gravitino.iceberg;
import com.google.common.collect.Lists;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import javax.inject.Singleton;
import javax.servlet.Servlet;
import org.apache.gravitino.Configs;
@@ -32,6 +33,8 @@ import
org.apache.gravitino.iceberg.service.IcebergCatalogWrapperManager;
import org.apache.gravitino.iceberg.service.IcebergExceptionMapper;
import org.apache.gravitino.iceberg.service.IcebergObjectMapperProvider;
import
org.apache.gravitino.iceberg.service.authorization.IcebergRESTServerContext;
+import org.apache.gravitino.iceberg.service.cleanup.IcebergCleanupJobStore;
+import org.apache.gravitino.iceberg.service.cleanup.IcebergCleanupManager;
import
org.apache.gravitino.iceberg.service.dispatcher.IcebergNamespaceEventDispatcher;
import
org.apache.gravitino.iceberg.service.dispatcher.IcebergNamespaceHookDispatcher;
import
org.apache.gravitino.iceberg.service.dispatcher.IcebergNamespaceOperationDispatcher;
@@ -76,6 +79,7 @@ public class RESTService implements GravitinoAuxiliaryService
{
private IcebergCatalogWrapperManager icebergCatalogWrapperManager;
private IcebergMetricsManager icebergMetricsManager;
+ private Optional<IcebergCleanupManager> cleanupManager;
private IcebergConfigProvider configProvider;
private boolean auxMode;
@@ -119,9 +123,25 @@ public class RESTService implements
GravitinoAuxiliaryService {
skipAuthorizationForRestBackend,
icebergCatalogWrapperManager);
this.icebergMetricsManager = new IcebergMetricsManager(icebergConfig);
+ if (auxMode) {
+ // Async cleanup reuses the entity store's shared relational backend
(connection pool +
+ // per-backend SQL), which is only available when running embedded in
the Gravitino server
+ // (auxiliary mode). In standalone mode the cleanup manager stays empty
and purge requests
+ // fall back to synchronous purge.
+ this.cleanupManager =
+ Optional.of(
+ new IcebergCleanupManager(
+ new
IcebergCleanupJobStore(GravitinoEnv.getInstance().idGenerator()),
+ icebergConfig));
+ } else {
+ this.cleanupManager = Optional.empty();
+ LOG.info(
+ "Async Iceberg table cleanup is only available in auxiliary mode; "
+ + "purge requests with async mode will fall back to synchronous
purge.");
+ }
// Table: HookDispatcher -> EventDispatcher -> OperationExecutor
IcebergTableOperationDispatcher icebergTableOperationDispatcher =
- new IcebergTableOperationExecutor(icebergCatalogWrapperManager);
+ new IcebergTableOperationExecutor(icebergCatalogWrapperManager,
cleanupManager);
IcebergTableOperationDispatcher icebergTableEventDispatcher =
new IcebergTableEventDispatcher(icebergTableOperationDispatcher,
eventBus, metalakeName);
if (authorizationContext.isAuthorizationEnabled()) {
@@ -142,7 +162,7 @@ public class RESTService implements
GravitinoAuxiliaryService {
// Namespace: HookDispatcher -> EventDispatcher -> OperationExecutor
IcebergNamespaceOperationDispatcher namespaceOperationDispatcher =
- new IcebergNamespaceOperationExecutor(icebergCatalogWrapperManager);
+ new IcebergNamespaceOperationExecutor(icebergCatalogWrapperManager,
cleanupManager);
IcebergNamespaceOperationDispatcher icebergNamespaceEventDispatcher =
new IcebergNamespaceEventDispatcher(namespaceOperationDispatcher,
eventBus, metalakeName);
if (authorizationContext.isAuthorizationEnabled()) {
@@ -163,6 +183,8 @@ public class RESTService implements
GravitinoAuxiliaryService {
}
bind(icebergCatalogWrapperManager).to(IcebergCatalogWrapperManager.class).ranked(1);
bind(icebergMetricsManager).to(IcebergMetricsManager.class).ranked(1);
+ cleanupManager.ifPresent(
+ manager ->
bind(manager).to(IcebergCleanupManager.class).ranked(1));
bind(icebergTableDispatcher).to(IcebergTableOperationDispatcher.class).ranked(1);
bind(icebergViewDispatcher).to(IcebergViewOperationDispatcher.class).ranked(1);
bind(icebergNamespaceDispatcher)
@@ -198,11 +220,15 @@ public class RESTService implements
GravitinoAuxiliaryService {
@Override
public void serviceStart() {
icebergMetricsManager.start();
+ cleanupManager.ifPresent(IcebergCleanupManager::start);
if (server != null) {
try {
server.start();
LOG.info("Iceberg REST service started");
} catch (Exception e) {
+ // Stop the components we already started so they don't outlive a
failed startup.
+ cleanupManager.ifPresent(IcebergCleanupManager::close);
+ icebergMetricsManager.close();
throw new RuntimeException(e);
}
}
@@ -223,6 +249,7 @@ public class RESTService implements
GravitinoAuxiliaryService {
if (icebergMetricsManager != null) {
icebergMetricsManager.close();
}
+ cleanupManager.ifPresent(IcebergCleanupManager::close);
}
public void join() {
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/IcebergCleanupJobStore.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/IcebergCleanupJobStore.java
index 6f715ba1d9..113a6e6a3a 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/IcebergCleanupJobStore.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/IcebergCleanupJobStore.java
@@ -23,9 +23,12 @@ import com.google.common.annotations.VisibleForTesting;
import java.util.List;
import java.util.Optional;
import
org.apache.gravitino.iceberg.service.cleanup.mapper.IcebergCleanupJobMapper;
+import
org.apache.gravitino.iceberg.service.cleanup.mapper.provider.IcebergCleanupMapperPackageProvider;
import org.apache.gravitino.iceberg.service.cleanup.po.IcebergCleanupJobPO;
import org.apache.gravitino.storage.IdGenerator;
+import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
import org.apache.gravitino.storage.relational.utils.SessionUtils;
+import org.apache.ibatis.session.Configuration;
/**
* Persistence for {@code iceberg_cleanup_job}, layered on the Gravitino
entity store's shared
@@ -47,6 +50,30 @@ public class IcebergCleanupJobStore {
*/
public IcebergCleanupJobStore(IdGenerator idGenerator) {
this.idGenerator = idGenerator;
+ registerMappers();
+ }
+
+ /**
+ * Registers the cleanup mappers into the entity store's shared MyBatis
configuration.
+ *
+ * <p>Gravitino core builds the shared {@code SqlSessionFactory} during
startup, before any
+ * auxiliary service is loaded, and in deploy mode the iceberg-rest-server
runs in an isolated
+ * auxiliary-service class loader that core cannot see into. Core therefore
cannot discover this
+ * module's mappers, and without them every cleanup query fails with {@code
BindingException}. We
+ * register them here instead, lazily, from within the class loader that can
see them the first
+ * time a store is created. The {@code hasMapper} guard keeps it idempotent
across stores and
+ * threads.
+ */
+ private static void registerMappers() {
+ Configuration configuration =
+
SqlSessionFactoryHelper.getInstance().getSqlSessionFactory().getConfiguration();
+ for (Class<?> mapper : new
IcebergCleanupMapperPackageProvider().getMapperClasses()) {
+ synchronized (configuration) {
+ if (!configuration.hasMapper(mapper)) {
+ configuration.addMapper(mapper);
+ }
+ }
+ }
}
/**
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/mapper/provider/IcebergCleanupMapperPackageProvider.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/mapper/provider/IcebergCleanupMapperPackageProvider.java
index cdfd225129..5235819add 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/mapper/provider/IcebergCleanupMapperPackageProvider.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/mapper/provider/IcebergCleanupMapperPackageProvider.java
@@ -25,9 +25,15 @@ import
org.apache.gravitino.iceberg.service.cleanup.mapper.IcebergCleanupJobMapp
import
org.apache.gravitino.storage.relational.mapper.provider.MapperPackageProvider;
/**
- * Registers the Iceberg async-cleanup mapper into the Gravitino entity
store's MyBatis
- * configuration via {@link java.util.ServiceLoader}, so the cleanup job store
reuses the shared
- * relational backend rather than opening its own JDBC connections.
+ * Lists the Iceberg async-cleanup mappers so the cleanup job store can
register them into the
+ * Gravitino entity store's shared MyBatis configuration and reuse the shared
relational backend
+ * rather than opening its own JDBC connections.
+ *
+ * <p>This implements {@link MapperPackageProvider} but is invoked directly
(not via {@link
+ * java.util.ServiceLoader}): in deploy mode the iceberg-rest-server runs in
an isolated
+ * auxiliary-service class loader that core's service lookup cannot see into,
so {@link
+ * org.apache.gravitino.iceberg.service.cleanup.IcebergCleanupJobStore}
consults this provider when
+ * it registers the mappers.
*/
public class IcebergCleanupMapperPackageProvider implements
MapperPackageProvider {
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergCleanupHelper.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergCleanupHelper.java
new file mode 100644
index 0000000000..b01f30eca0
--- /dev/null
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergCleanupHelper.java
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.gravitino.iceberg.service.dispatcher;
+
+import java.util.Optional;
+import org.apache.gravitino.GravitinoEnv;
+import org.apache.gravitino.NameIdentifier;
+import
org.apache.gravitino.iceberg.service.authorization.IcebergRESTServerContext;
+import org.apache.gravitino.iceberg.service.cleanup.IcebergCleanupManager;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.exceptions.AlreadyExistsException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** Request-path helpers shared by the table and namespace executors for async
table purge. */
+final class IcebergCleanupHelper {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(IcebergCleanupHelper.class);
+
+ private IcebergCleanupHelper() {}
+
+ /**
+ * Returns the catalog entity id for {@code catalogName}. The catalog is
already loaded for the
+ * current request, so this reads its in-memory entity without an extra
entity-store lookup.
+ */
+ static long catalogId(String catalogName) {
+ String metalake = IcebergRESTServerContext.getInstance().metalakeName();
+ return GravitinoEnv.getInstance()
+ .catalogManager()
+ .loadCatalogAndWrap(NameIdentifier.of(metalake, catalogName))
+ .catalog()
+ .entity()
+ .id();
+ }
+
+ /**
+ * Fails a create or register with {@code 409} while a cleanup job still
holds the identifier.
+ * Reusing the name before its files are gone would let the new table share
the old table's
+ * storage prefix. A name with no resolvable catalog entity cannot have a
cleanup job, so it stays
+ * usable.
+ */
+ static void rejectIfBeingPurged(
+ Optional<IcebergCleanupManager> cleanupManager,
+ String catalogName,
+ Namespace namespace,
+ String tableName) {
+ if (cleanupManager.isEmpty()) {
+ return;
+ }
+ long catalogId;
+ try {
+ catalogId = catalogId(catalogName);
+ } catch (RuntimeException e) {
+ LOG.warn("No catalog id for {}; skipping purge check", catalogName, e);
+ return;
+ }
+ if (cleanupManager.get().isNameOccupied(catalogId, namespace.toString(),
tableName)) {
+ throw new AlreadyExistsException(
+ "Table %s.%s is being purged; retry after cleanup completes",
namespace, tableName);
+ }
+ }
+}
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergNamespaceOperationExecutor.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergNamespaceOperationExecutor.java
index ebb9c6aac4..364ba55120 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergNamespaceOperationExecutor.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergNamespaceOperationExecutor.java
@@ -21,9 +21,11 @@ package org.apache.gravitino.iceberg.service.dispatcher;
import java.util.HashMap;
import java.util.Map;
+import java.util.Optional;
import org.apache.gravitino.auth.AuthConstants;
import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
import org.apache.gravitino.iceberg.service.IcebergCatalogWrapperManager;
+import org.apache.gravitino.iceberg.service.cleanup.IcebergCleanupManager;
import org.apache.gravitino.listener.api.event.IcebergRequestContext;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.rest.requests.CreateNamespaceRequest;
@@ -42,11 +44,14 @@ public class IcebergNamespaceOperationExecutor implements
IcebergNamespaceOperat
private static final Logger LOG =
LoggerFactory.getLogger(IcebergNamespaceOperationExecutor.class);
- private IcebergCatalogWrapperManager icebergCatalogWrapperManager;
+ private final IcebergCatalogWrapperManager icebergCatalogWrapperManager;
+ private final Optional<IcebergCleanupManager> cleanupManager;
public IcebergNamespaceOperationExecutor(
- IcebergCatalogWrapperManager icebergCatalogWrapperManager) {
+ IcebergCatalogWrapperManager icebergCatalogWrapperManager,
+ Optional<IcebergCleanupManager> cleanupManager) {
this.icebergCatalogWrapperManager = icebergCatalogWrapperManager;
+ this.cleanupManager = cleanupManager;
}
@Override
@@ -121,6 +126,9 @@ public class IcebergNamespaceOperationExecutor implements
IcebergNamespaceOperat
IcebergRequestContext context,
Namespace namespace,
RegisterTableRequest registerTableRequest) {
+ IcebergCleanupHelper.rejectIfBeingPurged(
+ cleanupManager, context.catalogName(), namespace,
registerTableRequest.name());
+
return icebergCatalogWrapperManager
.getCatalogWrapper(context.catalogName())
.registerTable(namespace, registerTableRequest,
context.requestCredentialVending());
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableOperationExecutor.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableOperationExecutor.java
index e32f2999f7..6307013315 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableOperationExecutor.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableOperationExecutor.java
@@ -27,13 +27,17 @@ import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.auth.AuthConstants;
import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
import org.apache.gravitino.credential.CredentialPrivilege;
+import org.apache.gravitino.iceberg.common.ops.IcebergCatalogWrapper;
import org.apache.gravitino.iceberg.common.utils.IcebergIdentifierUtils;
import org.apache.gravitino.iceberg.service.IcebergCatalogWrapperManager;
import
org.apache.gravitino.iceberg.service.authorization.IcebergRESTServerContext;
+import org.apache.gravitino.iceberg.service.cleanup.IcebergCleanupJob;
+import org.apache.gravitino.iceberg.service.cleanup.IcebergCleanupManager;
import org.apache.gravitino.listener.api.event.IcebergRequestContext;
import org.apache.gravitino.server.authorization.MetadataAuthzHelper;
import
org.apache.gravitino.server.authorization.expression.AuthorizationExpressionConstants;
import org.apache.gravitino.utils.HierarchicalSchemaUtil;
+import org.apache.iceberg.TableMetadata;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.rest.requests.CreateTableRequest;
@@ -51,15 +55,22 @@ public class IcebergTableOperationExecutor implements
IcebergTableOperationDispa
private static final Logger LOG =
LoggerFactory.getLogger(IcebergTableOperationExecutor.class);
- private IcebergCatalogWrapperManager icebergCatalogWrapperManager;
+ private final IcebergCatalogWrapperManager icebergCatalogWrapperManager;
+ private final Optional<IcebergCleanupManager> cleanupManager;
- public IcebergTableOperationExecutor(IcebergCatalogWrapperManager
icebergCatalogWrapperManager) {
+ public IcebergTableOperationExecutor(
+ IcebergCatalogWrapperManager icebergCatalogWrapperManager,
+ Optional<IcebergCleanupManager> cleanupManager) {
this.icebergCatalogWrapperManager = icebergCatalogWrapperManager;
+ this.cleanupManager = cleanupManager;
}
@Override
public LoadTableResponse createTable(
IcebergRequestContext context, Namespace namespace, CreateTableRequest
createTableRequest) {
+ IcebergCleanupHelper.rejectIfBeingPurged(
+ cleanupManager, context.catalogName(), namespace,
createTableRequest.name());
+
String authenticatedUser = context.userName();
if (!AuthConstants.ANONYMOUS_USER.equals(authenticatedUser)) {
String existingOwner =
createTableRequest.properties().get(IcebergConstants.OWNER);
@@ -110,15 +121,41 @@ public class IcebergTableOperationExecutor implements
IcebergTableOperationDispa
@Override
public void dropTable(
IcebergRequestContext context, TableIdentifier tableIdentifier, boolean
purgeRequested) {
- if (purgeRequested) {
- icebergCatalogWrapperManager
- .getCatalogWrapper(context.catalogName())
- .purgeTable(tableIdentifier);
- } else {
- icebergCatalogWrapperManager
- .getCatalogWrapper(context.catalogName())
- .dropTable(tableIdentifier);
+ IcebergCatalogWrapper wrapper =
+ icebergCatalogWrapperManager.getCatalogWrapper(context.catalogName());
+ if (!purgeRequested) {
+ wrapper.dropTable(tableIdentifier);
+ return;
+ }
+
+ // Async cleanup is opt-in per request and only wired in auxiliary mode;
otherwise purge inline.
+ if (!context.asyncPurge()) {
+ wrapper.purgeTable(tableIdentifier);
+ return;
}
+
+ // Async purge needs the cleanup manager, which only exists in auxiliary
mode. A request may
+ // still ask for async purge in standalone mode (empty manager); there is
no async engine to
+ // enqueue into, so we fall back to synchronous purge rather than fail the
request.
+ cleanupManager.ifPresentOrElse(
+ manager -> {
+ // Read the metadata location before dropping the catalog entry,
then enqueue the job. The
+ // job deletes only files reachable from this old metadata, so a
table recreated at the
+ // same name (with fresh metadata) is never touched.
+ TableMetadata metadata = wrapper.loadTableMetadata(tableIdentifier);
+ wrapper.dropTable(tableIdentifier);
+ manager.addJob(
+ new IcebergCleanupJob(
+ 0L,
+ IcebergCleanupHelper.catalogId(context.catalogName()),
+ tableIdentifier.namespace().toString(),
+ tableIdentifier.name(),
+ metadata.metadataFileLocation(),
+ wrapper.fileIOImpl(),
+ wrapper.fileIOProperties(),
+ context.userName()));
+ },
+ () -> wrapper.purgeTable(tableIdentifier));
}
@Override
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/listener/api/event/IcebergRequestContext.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/listener/api/event/IcebergRequestContext.java
index 4881257805..e10eb41734 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/listener/api/event/IcebergRequestContext.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/listener/api/event/IcebergRequestContext.java
@@ -28,6 +28,9 @@ import org.apache.gravitino.utils.PrincipalUtils;
/** The general request context information for Iceberg REST operations. */
public class IcebergRequestContext {
+ /** Header that opts a drop purge request into asynchronous file cleanup. */
+ public static final String ASYNC_PURGE_HEADER = "X-Gravitino-Async-Purge";
+
/**
* @deprecated Kept only for backward-compatibility and will be removed in
the next major release.
*/
@@ -113,6 +116,24 @@ public class IcebergRequestContext {
return httpHeaders;
}
+ /**
+ * Checks whether this request opted into asynchronous table purge.
+ *
+ * <p>Async purge is opt-in. Standard Iceberg clients send no header and
keep synchronous purge
+ * behavior; a client opts in with {@code X-Gravitino-Async-Purge: true}.
+ *
+ * @return true only when the async purge header explicitly says {@code true}
+ */
+ public boolean asyncPurge() {
+ for (Map.Entry<String, String> header : httpHeaders.entrySet()) {
+ // HTTP header names are case-insensitive; the value is matched exactly
as "true".
+ if (ASYNC_PURGE_HEADER.equalsIgnoreCase(header.getKey())) {
+ return "true".equals(header.getValue().trim());
+ }
+ }
+ return false;
+ }
+
/**
* Checks if the request is for credential vending.
*
diff --git
a/iceberg/iceberg-rest-server/src/main/resources/META-INF/services/org.apache.gravitino.storage.relational.mapper.provider.MapperPackageProvider
b/iceberg/iceberg-rest-server/src/main/resources/META-INF/services/org.apache.gravitino.storage.relational.mapper.provider.MapperPackageProvider
deleted file mode 100644
index 7ed1ef96c3..0000000000
---
a/iceberg/iceberg-rest-server/src/main/resources/META-INF/services/org.apache.gravitino.storage.relational.mapper.provider.MapperPackageProvider
+++ /dev/null
@@ -1,19 +0,0 @@
-#
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements. See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership. The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License. You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied. See the License for the
-# specific language governing permissions and limitations
-# under the License.
-#
-org.apache.gravitino.iceberg.service.cleanup.mapper.provider.IcebergCleanupMapperPackageProvider
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/integration/test/IcebergRESTAsyncPurgeIT.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/integration/test/IcebergRESTAsyncPurgeIT.java
new file mode 100644
index 0000000000..f798871b57
--- /dev/null
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/integration/test/IcebergRESTAsyncPurgeIT.java
@@ -0,0 +1,263 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.iceberg.integration.test;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Maps;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
+import org.apache.commons.io.FileUtils;
+import org.apache.gravitino.Catalog;
+import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
+import org.apache.gravitino.client.GravitinoMetalake;
+import org.apache.gravitino.iceberg.common.IcebergConfig;
+import org.apache.gravitino.integration.test.container.ContainerSuite;
+import org.apache.gravitino.integration.test.util.BaseIT;
+import org.apache.gravitino.integration.test.util.TestDatabaseName;
+import org.apache.gravitino.listener.api.event.IcebergRequestContext;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.AlreadyExistsException;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.io.PositionOutputStream;
+import org.apache.iceberg.rest.RESTCatalog;
+import org.apache.iceberg.types.Types;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * End-to-end test for async table purge. A table dropped with {@code
purgeRequested=true} plus the
+ * {@link IcebergRequestContext#ASYNC_PURGE_HEADER} should keep its files on
disk while the drop
+ * returns, block recreating the same name with {@code 409} until cleanup
finishes, and have its
+ * files deleted by the background worker.
+ *
+ * <p>It drives the server with the {@link RESTCatalog} client, not Spark:
Spark purges files
+ * client-side and sends {@code purgeRequested=false}, which skips the
server-side path. It uses the
+ * dynamic config provider over a PostgreSQL-backed {@code lakehouse-iceberg}
catalog with a local
+ * {@code file://} warehouse, since cleanup jobs are keyed by catalog id and
the files must be
+ * visible on disk.
+ */
+public class IcebergRESTAsyncPurgeIT extends BaseIT {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(IcebergRESTAsyncPurgeIT.class);
+ private static final String GRAVITINO_ICEBERG_REST_PREFIX =
"gravitino.iceberg-rest.";
+ private static final String METALAKE_NAME = "async_cleanup_metalake";
+ private static final String CATALOG_NAME = "iceberg_cleanup";
+ private static final String USER = "test";
+ private static final String DATABASE_NAME = "purge_db";
+ private static final String TABLE_NAME = "t_async";
+ // Poll interval large enough that the background worker provably cannot
claim the job during the
+ // synchronous in-flight assertions, yet small enough to keep the
eventual-cleanup wait short.
+ private static final int CLEANUP_POLL_INTERVAL_SECS = 5;
+
+ private static final Schema SCHEMA =
+ new Schema(
+ Types.NestedField.required(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "data", Types.StringType.get()));
+
+ private static final ContainerSuite CONTAINER_SUITE =
ContainerSuite.getInstance();
+
+ private Path warehouseDir;
+ private RESTCatalog restCatalog;
+
+ @BeforeAll
+ @Override
+ public void startIntegrationTest() throws Exception {
+
CONTAINER_SUITE.startPostgreSQLContainer(TestDatabaseName.PG_ICEBERG_ASYNC_CLEANUP_IT);
+ warehouseDir =
Files.createTempDirectory("gravitino-iceberg-async-cleanup");
+ ignoreIcebergAuxRestService = false;
+
+ // Simple authentication so the dynamic config provider can authenticate
to Gravitino and the
+ // REST client can identify itself; authorization stays disabled to keep
the test focused.
+ customConfigs.put("gravitino.authenticators", "simple");
+ customConfigs.put("SimpleAuthUserName", USER);
+ customConfigs.put(
+ GRAVITINO_ICEBERG_REST_PREFIX +
IcebergConstants.ICEBERG_REST_CATALOG_CONFIG_PROVIDER,
+ IcebergConstants.DYNAMIC_ICEBERG_CATALOG_CONFIG_PROVIDER_NAME);
+ customConfigs.put(
+ GRAVITINO_ICEBERG_REST_PREFIX + IcebergConstants.GRAVITINO_METALAKE,
METALAKE_NAME);
+ customConfigs.put(
+ GRAVITINO_ICEBERG_REST_PREFIX +
IcebergConstants.ICEBERG_REST_DEFAULT_DYNAMIC_CATALOG_NAME,
+ CATALOG_NAME);
+ customConfigs.put(
+ GRAVITINO_ICEBERG_REST_PREFIX +
IcebergConstants.GRAVITINO_SIMPLE_USERNAME, USER);
+ customConfigs.put(
+ GRAVITINO_ICEBERG_REST_PREFIX +
IcebergConfig.ASYNC_CLEANUP_POLL_INTERVAL_SECS.getKey(),
+ String.valueOf(CLEANUP_POLL_INTERVAL_SECS));
+
+ super.startIntegrationTest();
+ initMetalakeAndCatalog();
+ initRESTCatalog();
+ }
+
+ @AfterAll
+ @Override
+ public void stopIntegrationTest() throws IOException, InterruptedException {
+ if (restCatalog != null) {
+ try {
+ restCatalog.close();
+ } catch (IOException e) {
+ LOG.warn("Failed to close Iceberg REST catalog", e);
+ }
+ restCatalog = null;
+ }
+ try {
+ client.dropMetalake(METALAKE_NAME, true);
+ } catch (Exception e) {
+ LOG.warn("Failed to drop metalake {}", METALAKE_NAME, e);
+ }
+ super.stopIntegrationTest();
+ if (warehouseDir != null) {
+ FileUtils.deleteQuietly(warehouseDir.toFile());
+ }
+ }
+
+ @Test
+ void testAsyncPurgeEndToEnd() {
+ restCatalog.createNamespace(Namespace.of(DATABASE_NAME));
+ TableIdentifier identifier = TableIdentifier.of(DATABASE_NAME, TABLE_NAME);
+ Table table = restCatalog.createTable(identifier, SCHEMA);
+ appendDataFile(table);
+
+ Path tableDir = warehouseDir.resolve(DATABASE_NAME).resolve(TABLE_NAME);
+ Assertions.assertTrue(
+ countRegularFiles(tableDir) > 0,
+ "Table should have metadata and data files on disk before drop");
+
+ // Drop with purgeRequested=true; the async purge header routes this
through the cleanup
+ // manager.
+ restCatalog.dropTable(identifier, true);
+
+ // The worker polls every few seconds, so right after the drop returns the
files must still be
+ // on disk. Synchronous purge would have already deleted them, so this is
what proves the drop
+ // went through the async path.
+ Assertions.assertTrue(
+ countRegularFiles(tableDir) > 0,
+ "Async purge must not delete files synchronously during the drop
request");
+
+ // While the cleanup job is in flight, the dropped name is occupied and
cannot be recreated.
+ AlreadyExistsException occupied =
+ Assertions.assertThrows(
+ AlreadyExistsException.class, () ->
restCatalog.createTable(identifier, SCHEMA));
+ Assertions.assertTrue(
+ occupied.getMessage().contains("being purged"),
+ "Expected a 'being purged' conflict, got: " + occupied.getMessage());
+
+ // The background cleanup worker eventually deletes every file and
releases the name.
+ Awaitility.await()
+ .atMost(60, TimeUnit.SECONDS)
+ .pollInterval(500, TimeUnit.MILLISECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertEquals(
+ 0L,
+ countRegularFiles(tableDir),
+ "Background cleanup worker should have deleted all table
files"));
+ }
+
+ private void initMetalakeAndCatalog() {
+ GravitinoMetalake metalake = client.createMetalake(METALAKE_NAME, "", new
HashMap<>());
+ Map<String, String> props = Maps.newHashMap();
+ props.put(IcebergConstants.CATALOG_BACKEND, "jdbc");
+ props.put(
+ IcebergConstants.URI,
+ CONTAINER_SUITE
+ .getPostgreSQLContainer()
+ .getJdbcUrl(TestDatabaseName.PG_ICEBERG_ASYNC_CLEANUP_IT));
+ props.put(IcebergConstants.GRAVITINO_JDBC_DRIVER, "org.postgresql.Driver");
+ props.put(
+ IcebergConstants.GRAVITINO_JDBC_USER,
+ CONTAINER_SUITE.getPostgreSQLContainer().getUsername());
+ props.put(
+ IcebergConstants.GRAVITINO_JDBC_PASSWORD,
+ CONTAINER_SUITE.getPostgreSQLContainer().getPassword());
+ props.put("gravitino.bypass.jdbc.schema-version", "v1");
+ props.put(IcebergConstants.ICEBERG_JDBC_INITIALIZE, "true");
+ props.put(IcebergConstants.WAREHOUSE, warehouseDir.toUri().toString());
+
+ metalake.createCatalog(
+ CATALOG_NAME, Catalog.Type.RELATIONAL, "lakehouse-iceberg", "async
cleanup IT", props);
+ }
+
+ private void appendDataFile(Table table) {
+ // Purge deletes files by reachability, not content, so a placeholder file
is enough to exercise
+ // data-file cleanup alongside the manifests and metadata the commit
writes.
+ String dataPath = table.location() + "/data/async-purge-it-0.parquet";
+ OutputFile outputFile = table.io().newOutputFile(dataPath);
+ try (PositionOutputStream out = outputFile.create()) {
+ out.write(new byte[] {1, 2, 3, 4});
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to write placeholder data file " +
dataPath, e);
+ }
+ DataFile dataFile =
+ DataFiles.builder(PartitionSpec.unpartitioned())
+ .withPath(dataPath)
+ .withFormat(FileFormat.PARQUET)
+ .withFileSizeInBytes(4L)
+ .withRecordCount(1L)
+ .build();
+ table.newAppend().appendFile(dataFile).commit();
+ }
+
+ private void initRESTCatalog() {
+ String icebergRESTUri = getIcebergRestServiceUri();
+ LOG.info("Iceberg REST uri: {}", icebergRESTUri);
+ Map<String, String> props = new HashMap<>();
+ props.put(CatalogProperties.URI, icebergRESTUri);
+ props.put(CatalogProperties.CACHE_ENABLED, "false");
+ props.put("rest.auth.type", "basic");
+ props.put("rest.auth.basic.username", USER);
+ props.put("rest.auth.basic.password", "mock");
+ // Opt every request from this client into async purge.
+ props.put("header." + IcebergRequestContext.ASYNC_PURGE_HEADER, "true");
+ RESTCatalog catalog = new RESTCatalog();
+ catalog.setConf(new Configuration());
+ catalog.initialize("async_purge", ImmutableMap.copyOf(props));
+ restCatalog = catalog;
+ }
+
+ private long countRegularFiles(Path dir) {
+ if (!Files.exists(dir)) {
+ return 0L;
+ }
+ try (Stream<Path> paths = Files.walk(dir)) {
+ return paths.filter(Files::isRegularFile).count();
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to list files under " + dir, e);
+ }
+ }
+}
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/dispatcher/TestIcebergAsyncPurge.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/dispatcher/TestIcebergAsyncPurge.java
new file mode 100644
index 0000000000..3ee6bf6907
--- /dev/null
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/dispatcher/TestIcebergAsyncPurge.java
@@ -0,0 +1,260 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.gravitino.iceberg.service.dispatcher;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.Mockito.inOrder;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.Collections;
+import java.util.Optional;
+import org.apache.gravitino.GravitinoEnv;
+import org.apache.gravitino.auth.AuthConstants;
+import org.apache.gravitino.catalog.CatalogManager;
+import org.apache.gravitino.connector.BaseCatalog;
+import org.apache.gravitino.iceberg.service.CatalogWrapperForREST;
+import org.apache.gravitino.iceberg.service.IcebergCatalogWrapperManager;
+import
org.apache.gravitino.iceberg.service.authorization.IcebergRESTServerContext;
+import org.apache.gravitino.iceberg.service.cleanup.IcebergCleanupJob;
+import org.apache.gravitino.iceberg.service.cleanup.IcebergCleanupManager;
+import org.apache.gravitino.iceberg.service.provider.IcebergConfigProvider;
+import org.apache.gravitino.listener.api.event.IcebergRequestContext;
+import org.apache.gravitino.meta.CatalogEntity;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.AlreadyExistsException;
+import org.apache.iceberg.rest.requests.CreateTableRequest;
+import org.apache.iceberg.rest.requests.ImmutableRegisterTableRequest;
+import org.apache.iceberg.rest.requests.RegisterTableRequest;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.InOrder;
+import org.mockito.MockedStatic;
+
+/**
+ * Covers the async purge request path in the table and namespace executors:
how {@code dropTable}
+ * routes on the {@code X-Gravitino-Async-Purge} header, and how {@code
createTable} / {@code
+ * registerTable} are blocked while a cleanup job holds the identifier.
+ */
+class TestIcebergAsyncPurge {
+
+ private static final long CATALOG_ID = 42L;
+ private static final Namespace DB = Namespace.of("db");
+ private static final TableIdentifier TABLE = TableIdentifier.of("db", "t");
+ private static final Schema SCHEMA =
+ new Schema(Types.NestedField.required(1, "id", Types.LongType.get()));
+
+ @BeforeEach
+ void setUp() {
+ IcebergConfigProvider provider = mock(IcebergConfigProvider.class);
+ when(provider.getMetalakeName()).thenReturn("metalake");
+ when(provider.getDefaultCatalogName()).thenReturn("cat");
+ IcebergRESTServerContext.create(provider, false, false, true, null);
+ }
+
+ // --- dropTable routing ---
+
+ @Test
+ void testAsyncDropEnqueuesJob() {
+ CatalogWrapperForREST wrapper = mock(CatalogWrapperForREST.class);
+ IcebergCleanupManager cleanup = mock(IcebergCleanupManager.class);
+ TableMetadata metadata = mock(TableMetadata.class);
+
when(metadata.metadataFileLocation()).thenReturn("s3://b/db/t/metadata/0.json");
+ when(wrapper.loadTableMetadata(any())).thenReturn(metadata);
+ when(wrapper.fileIOImpl()).thenReturn("io");
+ when(wrapper.fileIOProperties()).thenReturn(Collections.emptyMap());
+
+ try (MockedStatic<GravitinoEnv> ignored = mockCatalogId()) {
+ tableExecutor(wrapper, Optional.of(cleanup)).dropTable(context(true),
TABLE, true);
+ }
+
+ InOrder ordered = inOrder(wrapper, cleanup);
+ ordered.verify(wrapper).loadTableMetadata(TABLE);
+ ordered.verify(wrapper).dropTable(TABLE);
+ ArgumentCaptor<IcebergCleanupJob> captor =
ArgumentCaptor.forClass(IcebergCleanupJob.class);
+ ordered.verify(cleanup).addJob(captor.capture());
+ IcebergCleanupJob job = captor.getValue();
+ Assertions.assertEquals(CATALOG_ID, job.catalogId());
+ Assertions.assertEquals("db", job.namespace());
+ Assertions.assertEquals("t", job.tableName());
+ Assertions.assertEquals("s3://b/db/t/metadata/0.json",
job.metadataLocation());
+ Assertions.assertEquals("io", job.fileIOImpl());
+ Assertions.assertEquals("alice", job.createdBy());
+ }
+
+ @Test
+ void testSyncPurgeByDefault() {
+ CatalogWrapperForREST wrapper = mock(CatalogWrapperForREST.class);
+ IcebergCleanupManager cleanup = mock(IcebergCleanupManager.class);
+
+ tableExecutor(wrapper, Optional.of(cleanup)).dropTable(context(false),
TABLE, true);
+
+ verify(wrapper).purgeTable(TABLE);
+ verify(cleanup, never()).addJob(any());
+ }
+
+ @Test
+ void testFallbackToSyncWhenDisabled() {
+ CatalogWrapperForREST wrapper = mock(CatalogWrapperForREST.class);
+
+ tableExecutor(wrapper, Optional.empty()).dropTable(context(true), TABLE,
true);
+
+ verify(wrapper).purgeTable(TABLE);
+ verify(wrapper, never()).loadTableMetadata(any());
+ }
+
+ @Test
+ void testPlainDropWhenNotPurge() {
+ CatalogWrapperForREST wrapper = mock(CatalogWrapperForREST.class);
+ IcebergCleanupManager cleanup = mock(IcebergCleanupManager.class);
+
+ tableExecutor(wrapper, Optional.of(cleanup)).dropTable(context(false),
TABLE, false);
+
+ verify(wrapper).dropTable(TABLE);
+ verify(cleanup, never()).addJob(any());
+ }
+
+ // --- create / register tombstone ---
+
+ @Test
+ void testCreateRejectedWhilePurging() {
+ CatalogWrapperForREST wrapper = mock(CatalogWrapperForREST.class);
+ IcebergCleanupManager cleanup = mock(IcebergCleanupManager.class);
+ when(cleanup.isNameOccupied(CATALOG_ID, "db", "t")).thenReturn(true);
+
+ try (MockedStatic<GravitinoEnv> ignored = mockCatalogId()) {
+ AlreadyExistsException e =
+ Assertions.assertThrows(
+ AlreadyExistsException.class,
+ () ->
+ tableExecutor(wrapper, Optional.of(cleanup))
+ .createTable(context(false), DB, createReq()));
+ Assertions.assertTrue(e.getMessage().contains("being purged"),
e.getMessage());
+ }
+ verify(wrapper, never()).createTable(any(), any(), anyBoolean());
+ }
+
+ @Test
+ void testCreateAllowedWhenNotPurging() {
+ CatalogWrapperForREST wrapper = mock(CatalogWrapperForREST.class);
+ IcebergCleanupManager cleanup = mock(IcebergCleanupManager.class);
+ when(cleanup.isNameOccupied(CATALOG_ID, "db", "t")).thenReturn(false);
+
+ try (MockedStatic<GravitinoEnv> ignored = mockCatalogId()) {
+ tableExecutor(wrapper, Optional.of(cleanup)).createTable(context(false),
DB, createReq());
+ }
+ verify(wrapper).createTable(any(), any(), anyBoolean());
+ }
+
+ @Test
+ void testRegisterRejectedWhilePurging() {
+ CatalogWrapperForREST wrapper = mock(CatalogWrapperForREST.class);
+ IcebergCleanupManager cleanup = mock(IcebergCleanupManager.class);
+ when(cleanup.isNameOccupied(CATALOG_ID, "db", "t")).thenReturn(true);
+
+ try (MockedStatic<GravitinoEnv> ignored = mockCatalogId()) {
+ AlreadyExistsException e =
+ Assertions.assertThrows(
+ AlreadyExistsException.class,
+ () ->
+ namespaceExecutor(wrapper, Optional.of(cleanup))
+ .registerTable(context(false), DB, registerReq()));
+ Assertions.assertTrue(e.getMessage().contains("being purged"),
e.getMessage());
+ }
+ verify(wrapper, never()).registerTable(any(), any(), anyBoolean());
+ }
+
+ @Test
+ void testRegisterAllowedWhenNotPurging() {
+ CatalogWrapperForREST wrapper = mock(CatalogWrapperForREST.class);
+ IcebergCleanupManager cleanup = mock(IcebergCleanupManager.class);
+ when(cleanup.isNameOccupied(CATALOG_ID, "db", "t")).thenReturn(false);
+
+ try (MockedStatic<GravitinoEnv> ignored = mockCatalogId()) {
+ namespaceExecutor(wrapper, Optional.of(cleanup))
+ .registerTable(context(false), DB, registerReq());
+ }
+ verify(wrapper).registerTable(any(), any(), anyBoolean());
+ }
+
+ // --- helpers ---
+
+ private IcebergTableOperationExecutor tableExecutor(
+ CatalogWrapperForREST wrapper, Optional<IcebergCleanupManager> cleanup) {
+ return new IcebergTableOperationExecutor(wrapperManager(wrapper), cleanup);
+ }
+
+ private IcebergNamespaceOperationExecutor namespaceExecutor(
+ CatalogWrapperForREST wrapper, Optional<IcebergCleanupManager> cleanup) {
+ return new IcebergNamespaceOperationExecutor(wrapperManager(wrapper),
cleanup);
+ }
+
+ private static IcebergCatalogWrapperManager
wrapperManager(CatalogWrapperForREST wrapper) {
+ IcebergCatalogWrapperManager manager =
mock(IcebergCatalogWrapperManager.class);
+ when(manager.getCatalogWrapper("cat")).thenReturn(wrapper);
+ return manager;
+ }
+
+ private static IcebergRequestContext context(boolean asyncPurge) {
+ IcebergRequestContext context = mock(IcebergRequestContext.class);
+ when(context.catalogName()).thenReturn("cat");
+ when(context.userName()).thenReturn(asyncPurge ? "alice" :
AuthConstants.ANONYMOUS_USER);
+ when(context.asyncPurge()).thenReturn(asyncPurge);
+ return context;
+ }
+
+ /** Stubs the request-thread catalog-id resolution to {@link #CATALOG_ID}. */
+ private static MockedStatic<GravitinoEnv> mockCatalogId() {
+ MockedStatic<GravitinoEnv> envStatic = mockStatic(GravitinoEnv.class);
+ GravitinoEnv env = mock(GravitinoEnv.class);
+ CatalogManager catalogManager = mock(CatalogManager.class);
+ CatalogManager.CatalogWrapper wrapper =
mock(CatalogManager.CatalogWrapper.class);
+ BaseCatalog<?> catalog = mock(BaseCatalog.class);
+ CatalogEntity entity = mock(CatalogEntity.class);
+ envStatic.when(GravitinoEnv::getInstance).thenReturn(env);
+ when(env.catalogManager()).thenReturn(catalogManager);
+ when(catalogManager.loadCatalogAndWrap(any())).thenReturn(wrapper);
+ when(wrapper.catalog()).thenReturn(catalog);
+ when(catalog.entity()).thenReturn(entity);
+ when(entity.id()).thenReturn(CATALOG_ID);
+ return envStatic;
+ }
+
+ private static CreateTableRequest createReq() {
+ return
CreateTableRequest.builder().withName("t").withSchema(SCHEMA).build();
+ }
+
+ private static RegisterTableRequest registerReq() {
+ return ImmutableRegisterTableRequest.builder()
+ .name("t")
+ .metadataLocation("s3://b/db/t/metadata/0.json")
+ .build();
+ }
+}
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/dispatcher/TestIcebergNamespaceOperationExecutor.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/dispatcher/TestIcebergNamespaceOperationExecutor.java
index f9cb7353ba..2e0e1b5a8d 100644
---
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/dispatcher/TestIcebergNamespaceOperationExecutor.java
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/dispatcher/TestIcebergNamespaceOperationExecutor.java
@@ -28,6 +28,7 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
+import java.util.Optional;
import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
import org.apache.gravitino.iceberg.service.CatalogWrapperForREST;
import org.apache.gravitino.iceberg.service.IcebergCatalogWrapperManager;
@@ -53,7 +54,7 @@ public class TestIcebergNamespaceOperationExecutor {
public void setUp() {
mockWrapperManager = mock(IcebergCatalogWrapperManager.class);
mockCatalogWrapper = mock(CatalogWrapperForREST.class);
- executor = new IcebergNamespaceOperationExecutor(mockWrapperManager);
+ executor = new IcebergNamespaceOperationExecutor(mockWrapperManager,
Optional.empty());
mockContext = mock(IcebergRequestContext.class);
when(mockContext.catalogName()).thenReturn("test_catalog");
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/dispatcher/TestIcebergTableOperationExecutor.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/dispatcher/TestIcebergTableOperationExecutor.java
index 65e271731d..9c5ebe158a 100644
---
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/dispatcher/TestIcebergTableOperationExecutor.java
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/dispatcher/TestIcebergTableOperationExecutor.java
@@ -27,6 +27,7 @@ import static org.mockito.Mockito.when;
import java.util.HashMap;
import java.util.Map;
+import java.util.Optional;
import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
import org.apache.gravitino.iceberg.service.CatalogWrapperForREST;
import org.apache.gravitino.iceberg.service.IcebergCatalogWrapperManager;
@@ -56,7 +57,7 @@ public class TestIcebergTableOperationExecutor {
public void setUp() {
mockWrapperManager = mock(IcebergCatalogWrapperManager.class);
mockCatalogWrapper = mock(CatalogWrapperForREST.class);
- executor = new IcebergTableOperationExecutor(mockWrapperManager);
+ executor = new IcebergTableOperationExecutor(mockWrapperManager,
Optional.empty());
mockContext = mock(IcebergRequestContext.class);
when(mockContext.catalogName()).thenReturn("test_catalog");
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/IcebergRestTestUtil.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/IcebergRestTestUtil.java
index 0f2c141800..8c572999b6 100644
---
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/IcebergRestTestUtil.java
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/IcebergRestTestUtil.java
@@ -27,6 +27,7 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.logging.Level;
import java.util.stream.Stream;
import javax.servlet.http.HttpServletRequest;
@@ -38,6 +39,7 @@ import
org.apache.gravitino.iceberg.service.IcebergExceptionMapper;
import org.apache.gravitino.iceberg.service.IcebergObjectMapperProvider;
import org.apache.gravitino.iceberg.service.IcebergRESTUtils;
import
org.apache.gravitino.iceberg.service.authorization.IcebergRESTServerContext;
+import org.apache.gravitino.iceberg.service.cleanup.IcebergCleanupManager;
import
org.apache.gravitino.iceberg.service.dispatcher.IcebergNamespaceEventDispatcher;
import
org.apache.gravitino.iceberg.service.dispatcher.IcebergNamespaceOperationDispatcher;
import
org.apache.gravitino.iceberg.service.dispatcher.IcebergNamespaceOperationExecutor;
@@ -146,7 +148,8 @@ public class IcebergRestTestUtil {
EventBus eventBus = new EventBus(eventListenerPlugins);
IcebergTableOperationExecutor icebergTableOperationExecutor =
- new IcebergTableOperationExecutor(icebergCatalogWrapperManager);
+ new IcebergTableOperationExecutor(
+ icebergCatalogWrapperManager,
Optional.of(mock(IcebergCleanupManager.class)));
IcebergTableEventDispatcher icebergTableEventDispatcher =
new IcebergTableEventDispatcher(
icebergTableOperationExecutor, eventBus,
configProvider.getMetalakeName());
@@ -156,7 +159,8 @@ public class IcebergRestTestUtil {
new IcebergViewEventDispatcher(
icebergViewOperationExecutor, eventBus,
configProvider.getMetalakeName());
IcebergNamespaceOperationExecutor icebergNamespaceOperationExecutor =
- new IcebergNamespaceOperationExecutor(icebergCatalogWrapperManager);
+ new IcebergNamespaceOperationExecutor(
+ icebergCatalogWrapperManager,
Optional.of(mock(IcebergCleanupManager.class)));
IcebergNamespaceEventDispatcher icebergNamespaceEventDispatcher =
new IcebergNamespaceEventDispatcher(
icebergNamespaceOperationExecutor, eventBus,
configProvider.getMetalakeName());
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/listener/api/event/TestIcebergRequestContext.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/listener/api/event/TestIcebergRequestContext.java
new file mode 100644
index 0000000000..eeff0219d2
--- /dev/null
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/listener/api/event/TestIcebergRequestContext.java
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.gravitino.listener.api.event;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.Collections;
+import java.util.Enumeration;
+import javax.servlet.http.HttpServletRequest;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class TestIcebergRequestContext {
+
+ @Test
+ void testNoHeaderIsSync() {
+ Assertions.assertFalse(new IcebergRequestContext(requestWithoutHeader(),
"cat").asyncPurge());
+ }
+
+ @Test
+ void testTrueHeaderIsAsync() {
+ Assertions.assertTrue(
+ new IcebergRequestContext(requestWithAsyncPurgeHeader(" true "),
"cat").asyncPurge());
+ }
+
+ @Test
+ void testFalseHeaderIsSync() {
+ Assertions.assertFalse(
+ new IcebergRequestContext(requestWithAsyncPurgeHeader("false"),
"cat").asyncPurge());
+ }
+
+ @Test
+ void testGarbageHeaderIsSync() {
+ Assertions.assertFalse(
+ new IcebergRequestContext(requestWithAsyncPurgeHeader("yes"),
"cat").asyncPurge());
+ }
+
+ @Test
+ void testUppercaseValueIsSync() {
+ // HTTP header values are case-sensitive; only the exact value "true" opts
in.
+ Assertions.assertFalse(
+ new IcebergRequestContext(requestWithAsyncPurgeHeader("True"),
"cat").asyncPurge());
+ }
+
+ @Test
+ void testHeaderNameIsCaseInsensitive() {
+ Assertions.assertTrue(
+ new IcebergRequestContext(requestWithHeader("x-gravitino-async-purge",
"true"), "cat")
+ .asyncPurge());
+ }
+
+ private static HttpServletRequest requestWithoutHeader() {
+ HttpServletRequest request = mock(HttpServletRequest.class);
+ when(request.getRemoteHost()).thenReturn("localhost");
+ when(request.getHeaderNames()).thenReturn(Collections.emptyEnumeration());
+ return request;
+ }
+
+ private static HttpServletRequest requestWithAsyncPurgeHeader(String value) {
+ return requestWithHeader(IcebergRequestContext.ASYNC_PURGE_HEADER, value);
+ }
+
+ private static HttpServletRequest requestWithHeader(String name, String
value) {
+ HttpServletRequest request = mock(HttpServletRequest.class);
+ Enumeration<String> headerNames =
Collections.enumeration(Collections.singleton(name));
+ when(request.getRemoteHost()).thenReturn("localhost");
+ when(request.getHeaderNames()).thenReturn(headerNames);
+ when(request.getHeader(name)).thenReturn(value);
+ return request;
+ }
+}
