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 76fe5d3b91 [#11265] feat(iceberg-rest): add async cleanup persistence
layer (#11266)
76fe5d3b91 is described below
commit 76fe5d3b91b3683415982b49e02a52a92ba95515
Author: roryqi <[email protected]>
AuthorDate: Fri May 29 21:35:12 2026 +0800
[#11265] feat(iceberg-rest): add async cleanup persistence layer (#11266)
### What changes were proposed in this pull request?
First of three stacked PRs for async Iceberg hard deletion - the
**persistence foundation only**:
- New IcebergConfig keys for the async-cleanup worker/retry knobs.
- iceberg_cleanup_job table schema + 1.2.0->1.3.0 migrations for
H2/MySQL/PostgreSQL.
- IcebergCleanupJob value object with a nested State enum.
- IcebergCleanupJobStore - add jobs, take runnable jobs with heartbeat
CAS, mark progress, and delete old finished jobs - with its MyBatis
mapper/PO/SQL provider.
- core test-jar (testArtifacts) wiring so the store tests can reuse the
relational entity-store backend.
Nothing is wired into the REST drop flow yet; that comes in the
follow-up PRs (manager engine, then REST integration).
### Why are the changes needed?
Inline cleanup of a large Iceberg table blocks the drop request and
risks client timeouts, and a crash mid-deletion leaves no record to
resume. A durable, persisted job store is the foundation for an opt-in
async cleanup. Splitting it out keeps each PR reviewable.
Fix: #11265
### Does this PR introduce _any_ user-facing change?
Adds new Iceberg REST config property keys (opt-in, unused until later
PRs) and a new iceberg_cleanup_job table in the 1.3.0 schema/upgrade
scripts. No API or default-behavior change.
### How was this patch tested?
New unit tests TestIcebergCleanupJob and TestIcebergCleanupJobStore (the
latter runs the relational backend matrix via
AbstractIcebergCleanupJobStoreBackendTest). Verified
:iceberg:iceberg-common:testClasses and
:iceberg:iceberg-rest-server:testClasses compile.
---
core/build.gradle.kts | 13 ++
.../gravitino/iceberg/common/IcebergConfig.java | 56 ++++++
.../iceberg/common/TestIcebergConfig.java | 12 ++
iceberg/iceberg-rest-server/build.gradle.kts | 1 +
.../iceberg/service/cleanup/IcebergCleanupJob.java | 100 ++++++++++
.../service/cleanup/IcebergCleanupJobStore.java | 209 +++++++++++++++++++++
.../cleanup/mapper/IcebergCleanupJobMapper.java | 84 +++++++++
.../IcebergCleanupJobSQLProviderFactory.java | 110 +++++++++++
.../IcebergCleanupMapperPackageProvider.java | 38 ++++
.../base/IcebergCleanupJobBaseSQLProvider.java | 185 ++++++++++++++++++
.../service/cleanup/po/IcebergCleanupJobPO.java | 122 ++++++++++++
...elational.mapper.provider.MapperPackageProvider | 19 ++
.../AbstractIcebergCleanupJobStoreBackendTest.java | 146 ++++++++++++++
.../service/cleanup/TestIcebergCleanupJob.java | 53 ++++++
.../integration/test/util/TestDatabaseName.java | 2 +
scripts/h2/schema-1.3.0-h2.sql | 19 ++
scripts/h2/upgrade-1.2.0-to-1.3.0-h2.sql | 19 ++
scripts/mysql/schema-1.3.0-mysql.sql | 19 ++
scripts/mysql/upgrade-1.2.0-to-1.3.0-mysql.sql | 19 ++
scripts/postgresql/schema-1.3.0-postgresql.sql | 32 ++++
.../upgrade-1.2.0-to-1.3.0-postgresql.sql | 32 ++++
21 files changed, 1290 insertions(+)
diff --git a/core/build.gradle.kts b/core/build.gradle.kts
index e92a7f52ab..a8511341f0 100644
--- a/core/build.gradle.kts
+++ b/core/build.gradle.kts
@@ -90,6 +90,19 @@ dependencies {
jcstressImplementation(libs.aspectj.aspectjrt)
}
+val testJar by tasks.registering(Jar::class) {
+ archiveClassifier.set("tests")
+ from(sourceSets["test"].output)
+}
+
+configurations {
+ create("testArtifacts")
+}
+
+artifacts {
+ add("testArtifacts", testJar)
+}
+
tasks.test {
val testMode = project.properties["testMode"] as? String ?: "embedded"
if (testMode == "embedded") {
diff --git
a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/IcebergConfig.java
b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/IcebergConfig.java
index f5ee8dd512..aeba5f01a8 100644
---
a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/IcebergConfig.java
+++
b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/IcebergConfig.java
@@ -319,6 +319,62 @@ public class IcebergConfig extends Config implements
OverwriteDefaultConfig {
.checkValue(value -> value > 0,
ConfigConstants.POSITIVE_NUMBER_ERROR_MSG)
.createWithDefault(60);
+ public static final ConfigEntry<Integer> ASYNC_CLEANUP_WORKER_THREADS =
+ new ConfigBuilder("async-cleanup.worker-threads")
+ .doc("Worker pool size per server (concurrent async cleanup jobs).")
+ .version(ConfigConstants.VERSION_1_3_0)
+ .intConf()
+ .checkValue(value -> value > 0,
ConfigConstants.POSITIVE_NUMBER_ERROR_MSG)
+ .createWithDefault(2);
+
+ public static final ConfigEntry<Integer> ASYNC_CLEANUP_DELETE_THREADS =
+ new ConfigBuilder("async-cleanup.delete-threads")
+ .doc("Server-wide file-delete pool size, shared across all cleanup
jobs.")
+ .version(ConfigConstants.VERSION_1_3_0)
+ .intConf()
+ .checkValue(value -> value > 0,
ConfigConstants.POSITIVE_NUMBER_ERROR_MSG)
+ .createWithDefault(4);
+
+ public static final ConfigEntry<Integer> ASYNC_CLEANUP_DELETE_BATCH_SIZE =
+ new ConfigBuilder("async-cleanup.delete-batch-size")
+ .doc("Files per bulk-delete batch handed to the delete executor.")
+ .version(ConfigConstants.VERSION_1_3_0)
+ .intConf()
+ .checkValue(value -> value > 0,
ConfigConstants.POSITIVE_NUMBER_ERROR_MSG)
+ .createWithDefault(1000);
+
+ public static final ConfigEntry<Integer> ASYNC_CLEANUP_POLL_INTERVAL_SECS =
+ new ConfigBuilder("async-cleanup.poll-interval-secs")
+ .doc("Worker poll interval in seconds; also the retry interval.")
+ .version(ConfigConstants.VERSION_1_3_0)
+ .intConf()
+ .checkValue(value -> value > 0,
ConfigConstants.POSITIVE_NUMBER_ERROR_MSG)
+ .createWithDefault(5);
+
+ public static final ConfigEntry<Integer>
ASYNC_CLEANUP_HEARTBEAT_TIMEOUT_SECS =
+ new ConfigBuilder("async-cleanup.heartbeat-timeout-secs")
+ .doc("Age in seconds after which a stale-heartbeat job can be taken
over by a worker.")
+ .version(ConfigConstants.VERSION_1_3_0)
+ .intConf()
+ .checkValue(value -> value > 0,
ConfigConstants.POSITIVE_NUMBER_ERROR_MSG)
+ .createWithDefault(300);
+
+ public static final ConfigEntry<Integer> ASYNC_CLEANUP_MAX_ATTEMPTS =
+ new ConfigBuilder("async-cleanup.max-attempts")
+ .doc("Number of attempts before a cleanup job is marked FAILED.")
+ .version(ConfigConstants.VERSION_1_3_0)
+ .intConf()
+ .checkValue(value -> value > 0,
ConfigConstants.POSITIVE_NUMBER_ERROR_MSG)
+ .createWithDefault(5);
+
+ public static final ConfigEntry<Integer> ASYNC_CLEANUP_RETENTION_HOURS =
+ new ConfigBuilder("async-cleanup.retention-hours")
+ .doc("How long finished (SUCCEEDED/FAILED) cleanup rows are retained
before pruning.")
+ .version(ConfigConstants.VERSION_1_3_0)
+ .intConf()
+ .checkValue(value -> value > 0,
ConfigConstants.POSITIVE_NUMBER_ERROR_MSG)
+ .createWithDefault(720);
+
public String getJdbcDriver() {
return get(JDBC_DRIVER);
}
diff --git
a/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/TestIcebergConfig.java
b/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/TestIcebergConfig.java
index 1bd47fcb16..ad8faaa236 100644
---
a/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/TestIcebergConfig.java
+++
b/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/TestIcebergConfig.java
@@ -87,4 +87,16 @@ public class TestIcebergConfig {
Assertions.assertFalse(
icebergConfigWithNewKey.get(IcebergConfig.ICEBERG_REST_DISABLE_REST_AUTHZ));
}
+
+ @Test
+ public void testAsyncCleanupDefaults() {
+ IcebergConfig config = new IcebergConfig(ImmutableMap.of());
+ Assertions.assertEquals(2,
config.get(IcebergConfig.ASYNC_CLEANUP_WORKER_THREADS));
+ Assertions.assertEquals(4,
config.get(IcebergConfig.ASYNC_CLEANUP_DELETE_THREADS));
+ Assertions.assertEquals(1000,
config.get(IcebergConfig.ASYNC_CLEANUP_DELETE_BATCH_SIZE));
+ Assertions.assertEquals(5,
config.get(IcebergConfig.ASYNC_CLEANUP_POLL_INTERVAL_SECS));
+ Assertions.assertEquals(300,
config.get(IcebergConfig.ASYNC_CLEANUP_HEARTBEAT_TIMEOUT_SECS));
+ Assertions.assertEquals(5,
config.get(IcebergConfig.ASYNC_CLEANUP_MAX_ATTEMPTS));
+ Assertions.assertEquals(720,
config.get(IcebergConfig.ASYNC_CLEANUP_RETENTION_HOURS));
+ }
}
diff --git a/iceberg/iceberg-rest-server/build.gradle.kts
b/iceberg/iceberg-rest-server/build.gradle.kts
index 60b0980987..59485de5ca 100644
--- a/iceberg/iceberg-rest-server/build.gradle.kts
+++ b/iceberg/iceberg-rest-server/build.gradle.kts
@@ -82,6 +82,7 @@ dependencies {
testImplementation(project(":bundles:iceberg-aws-bundle"))
testImplementation(project(":bundles:iceberg-gcp-bundle"))
testImplementation(project(":bundles:iceberg-azure-bundle"))
+ testImplementation(project(":core", "testArtifacts"))
testImplementation(project(":integration-test-common", "testArtifacts"))
testImplementation(project(":server"))
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/IcebergCleanupJob.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/IcebergCleanupJob.java
new file mode 100644
index 0000000000..5828d3497b
--- /dev/null
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/IcebergCleanupJob.java
@@ -0,0 +1,100 @@
+/*
+ * 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.cleanup;
+
+import com.google.common.collect.ImmutableMap;
+import java.util.Map;
+import lombok.Getter;
+import lombok.experimental.Accessors;
+
+/**
+ * Immutable description of a table to cleanup. Carries exactly the fields the
request path supplies
+ * at enqueue and the worker reads back to delete files. Mutable progress
(state, attempts,
+ * heartbeat) lives only in the {@code iceberg_cleanup_job} row, managed by
{@link
+ * IcebergCleanupJobStore}; the {@link State} enum here names those row states.
+ */
+@Getter
+@Accessors(fluent = true)
+public class IcebergCleanupJob {
+
+ /** Lifecycle states of an {@code iceberg_cleanup_job} row. */
+ public enum State {
+ /** Awaiting a worker; also the state a transiently-failed job returns to.
*/
+ PENDING,
+
+ /** Taken by a worker that is actively deleting files. */
+ RUNNING,
+
+ /** Every reachable file was deleted or already gone. */
+ SUCCEEDED,
+
+ /** Retries exhausted or a non-retryable failure; some files may remain
undeleted. */
+ FAILED;
+
+ /**
+ * Whether the job has finished, i.e. reached a final state that a worker
no longer acts on.
+ *
+ * @return {@code true} for {@link #SUCCEEDED} and {@link #FAILED}
+ */
+ public boolean isFinished() {
+ return this == SUCCEEDED || this == FAILED;
+ }
+ }
+
+ private final long id;
+ private final long catalogId;
+ private final String namespace;
+ private final String tableName;
+ private final String metadataLocation;
+ private final String fileIOImpl;
+ private final Map<String, String> fileIOProperties;
+ private final String createdBy;
+
+ /**
+ * Creates a cleanup job description.
+ *
+ * @param id row id, or {@code 0} before persistence
+ * @param catalogId globally unique id of the owning catalog; stable across
catalog rename
+ * @param namespace table namespace (dotted)
+ * @param tableName table name
+ * @param metadataLocation the dropped table's {@code metadata.json} location
+ * @param fileIOImpl FileIO implementation class to reconstruct in the worker
+ * @param fileIOProperties properties to reconstruct the FileIO, snapshotted
at enqueue
+ * @param createdBy principal that requested the drop
+ */
+ public IcebergCleanupJob(
+ long id,
+ long catalogId,
+ String namespace,
+ String tableName,
+ String metadataLocation,
+ String fileIOImpl,
+ Map<String, String> fileIOProperties,
+ String createdBy) {
+ this.id = id;
+ this.catalogId = catalogId;
+ this.namespace = namespace;
+ this.tableName = tableName;
+ this.metadataLocation = metadataLocation;
+ this.fileIOImpl = fileIOImpl;
+ this.fileIOProperties = ImmutableMap.copyOf(fileIOProperties);
+ this.createdBy = createdBy;
+ }
+}
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
new file mode 100644
index 0000000000..2bd3cd10a7
--- /dev/null
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/IcebergCleanupJobStore.java
@@ -0,0 +1,209 @@
+/*
+ * 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.cleanup;
+
+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.po.IcebergCleanupJobPO;
+import org.apache.gravitino.storage.IdGenerator;
+import org.apache.gravitino.storage.relational.utils.SessionUtils;
+
+/**
+ * Persistence for {@code iceberg_cleanup_job}, layered on the Gravitino
entity store's shared
+ * relational backend. Async cleanup reuses the entity store's connection
pool, transaction
+ * management, and per-backend SQL dispatch instead of opening its own JDBC
connections. Row ids and
+ * timestamps are supplied by the application, keeping the SQL portable across
H2, MySQL, and
+ * PostgreSQL.
+ */
+public class IcebergCleanupJobStore {
+
+ private static final int MAX_ERROR_LENGTH = 2048;
+
+ private final IdGenerator idGenerator;
+
+ /**
+ * Creates a cleanup job store.
+ *
+ * @param idGenerator generator for new row ids
+ */
+ public IcebergCleanupJobStore(IdGenerator idGenerator) {
+ this.idGenerator = idGenerator;
+ }
+
+ /**
+ * Persists a new PENDING job.
+ *
+ * @param job job to persist
+ * @return generated id
+ */
+ public long addJob(IcebergCleanupJob job) {
+ long id = idGenerator.nextId();
+ long now = System.currentTimeMillis();
+ IcebergCleanupJobPO po = IcebergCleanupJobPO.fromCleanupJob(job, id, now);
+ SessionUtils.doWithCommit(IcebergCleanupJobMapper.class, mapper ->
mapper.insertCleanupJob(po));
+ return id;
+ }
+
+ /**
+ * Scans a small candidate window and takes the first available row via
compare-and-swap.
+ *
+ * @param now current epoch millis, written as the initial heartbeat
+ * @param heartbeatTimeoutMs age past which a RUNNING heartbeat is stale
+ * @param window max candidates to consider
+ * @return the taken job, or {@link Optional#empty()} if nothing was
available
+ */
+ public Optional<IcebergCleanupJob> takePendingJob(long now, long
heartbeatTimeoutMs, int window) {
+ long heartbeatExpiry = now - heartbeatTimeoutMs;
+ List<IcebergCleanupJobPO> candidates =
+ SessionUtils.getWithoutCommit(
+ IcebergCleanupJobMapper.class,
+ mapper -> mapper.selectCandidateJobs(heartbeatExpiry, window));
+ for (IcebergCleanupJobPO po : candidates) {
+ long id = po.getId();
+ int marked =
+ SessionUtils.doWithCommitAndFetchResult(
+ IcebergCleanupJobMapper.class,
+ mapper -> mapper.markRunning(id, now, heartbeatExpiry));
+ if (marked == 1) {
+ // The claim only flips mutable columns (state, heartbeat_at,
updated_at); everything
+ // toCleanupJob reads was fixed at enqueue, so the candidate snapshot
is still accurate.
+ return Optional.of(po.toCleanupJob());
+ }
+ }
+ return Optional.empty();
+ }
+
+ /**
+ * Marks a RUNNING job SUCCEEDED.
+ *
+ * @param id job id
+ * @return {@code true} iff the row was still RUNNING and was updated (i.e.
the caller still owned
+ * the job)
+ */
+ public boolean markSucceeded(long id) {
+ long now = System.currentTimeMillis();
+ return SessionUtils.doWithCommitAndFetchResult(
+ IcebergCleanupJobMapper.class,
+ mapper -> mapper.markFinished(id,
IcebergCleanupJob.State.SUCCEEDED.name(), null, now))
+ > 0;
+ }
+
+ /**
+ * Marks a RUNNING job FAILED immediately.
+ *
+ * @param id job id
+ * @param reason failure text
+ * @return {@code true} iff the row was still RUNNING and was updated (i.e.
the caller still owned
+ * the job)
+ */
+ public boolean markFailed(long id, String reason) {
+ long now = System.currentTimeMillis();
+ String err = truncate(reason);
+ return SessionUtils.doWithCommitAndFetchResult(
+ IcebergCleanupJobMapper.class,
+ mapper -> mapper.markFinished(id,
IcebergCleanupJob.State.FAILED.name(), err, now))
+ > 0;
+ }
+
+ /**
+ * Records a transient failure: {@code attempts++}, then FAILED at the
ceiling else PENDING.
+ *
+ * @param id job id
+ * @param reason failure text
+ * @param maxAttempts ceiling from config
+ * @return {@code true} iff the row was still RUNNING and was updated (i.e.
the caller still owned
+ * the job)
+ */
+ public boolean recordFailure(long id, String reason, int maxAttempts) {
+ long now = System.currentTimeMillis();
+ String err = truncate(reason);
+ return SessionUtils.doWithCommitAndFetchResult(
+ IcebergCleanupJobMapper.class,
+ mapper -> mapper.recordFailure(id, err, maxAttempts, now))
+ > 0;
+ }
+
+ /**
+ * Refreshes a heartbeat with compare-and-swap ownership check.
+ *
+ * @param id job id
+ * @param lastHeartbeat previous heartbeat value
+ * @param now new heartbeat value
+ * @return {@code true} iff the row was still owned by the caller
+ */
+ public boolean heartbeat(long id, long lastHeartbeat, long now) {
+ return SessionUtils.doWithCommitAndFetchResult(
+ IcebergCleanupJobMapper.class, mapper -> mapper.heartbeat(id,
lastHeartbeat, now))
+ > 0;
+ }
+
+ /**
+ * Finds the id of an unfinished (PENDING or RUNNING) cleanup job for the
identifier, if any.
+ *
+ * @param catalogId globally unique id of the owning catalog
+ * @param namespace table namespace
+ * @param table table name
+ * @return the unfinished job id, or {@link Optional#empty()} if none exists
+ */
+ public Optional<Long> findUnfinishedJobId(long catalogId, String namespace,
String table) {
+ return Optional.ofNullable(
+ SessionUtils.getWithoutCommit(
+ IcebergCleanupJobMapper.class,
+ mapper -> mapper.selectUnfinishedJobId(catalogId, namespace,
table)));
+ }
+
+ /**
+ * Deletes finished (SUCCEEDED or FAILED) jobs whose last update predates
the timeline.
+ *
+ * @param legacyTimeline cutoff epoch millis; rows updated before this are
removed
+ * @return rows deleted
+ */
+ public int deleteFinishedJobsByLegacyTimeline(long legacyTimeline) {
+ return SessionUtils.doWithCommitAndFetchResult(
+ IcebergCleanupJobMapper.class,
+ mapper -> mapper.deleteFinishedJobsByLegacyTimeline(legacyTimeline));
+ }
+
+ /**
+ * Reads a job state for tests.
+ *
+ * @param id job id
+ * @return its current state
+ * @throws IllegalStateException if the row is gone
+ */
+ @VisibleForTesting
+ IcebergCleanupJob.State stateOf(long id) {
+ String state =
+ SessionUtils.getWithoutCommit(
+ IcebergCleanupJobMapper.class, mapper -> mapper.selectState(id));
+ if (state == null) {
+ throw new IllegalStateException("No cleanup job " + id);
+ }
+ return IcebergCleanupJob.State.valueOf(state);
+ }
+
+ private static String truncate(String value) {
+ return value == null || value.length() <= MAX_ERROR_LENGTH
+ ? value
+ : value.substring(0, MAX_ERROR_LENGTH);
+ }
+}
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/mapper/IcebergCleanupJobMapper.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/mapper/IcebergCleanupJobMapper.java
new file mode 100644
index 0000000000..12c981fb2d
--- /dev/null
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/mapper/IcebergCleanupJobMapper.java
@@ -0,0 +1,84 @@
+/*
+ * 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.cleanup.mapper;
+
+import java.util.List;
+import org.apache.gravitino.iceberg.service.cleanup.po.IcebergCleanupJobPO;
+import org.apache.ibatis.annotations.DeleteProvider;
+import org.apache.ibatis.annotations.InsertProvider;
+import org.apache.ibatis.annotations.Param;
+import org.apache.ibatis.annotations.SelectProvider;
+import org.apache.ibatis.annotations.UpdateProvider;
+
+/**
+ * MyBatis mapper for the {@code iceberg_cleanup_job} table. SQL is supplied
per backend by {@link
+ * IcebergCleanupJobSQLProviderFactory} and executed through the Gravitino
entity store's shared
+ * {@code SqlSessionFactory}, so async cleanup reuses the relational backend's
connection pool and
+ * multi-backend handling instead of opening its own JDBC connections.
+ */
+public interface IcebergCleanupJobMapper {
+
+ String TABLE_NAME = "iceberg_cleanup_job";
+
+ @InsertProvider(type = IcebergCleanupJobSQLProviderFactory.class, method =
"insertCleanupJob")
+ void insertCleanupJob(@Param("po") IcebergCleanupJobPO po);
+
+ @SelectProvider(type = IcebergCleanupJobSQLProviderFactory.class, method =
"selectCandidateJobs")
+ List<IcebergCleanupJobPO> selectCandidateJobs(
+ @Param("heartbeatExpiry") long heartbeatExpiry, @Param("window") int
window);
+
+ @UpdateProvider(type = IcebergCleanupJobSQLProviderFactory.class, method =
"markRunning")
+ int markRunning(
+ @Param("id") long id, @Param("now") long now, @Param("heartbeatExpiry")
long heartbeatExpiry);
+
+ @UpdateProvider(type = IcebergCleanupJobSQLProviderFactory.class, method =
"markFinished")
+ int markFinished(
+ @Param("id") long id,
+ @Param("state") String state,
+ @Param("reason") String reason,
+ @Param("now") long now);
+
+ @UpdateProvider(type = IcebergCleanupJobSQLProviderFactory.class, method =
"recordFailure")
+ int recordFailure(
+ @Param("id") long id,
+ @Param("reason") String reason,
+ @Param("maxAttempts") int maxAttempts,
+ @Param("now") long now);
+
+ @UpdateProvider(type = IcebergCleanupJobSQLProviderFactory.class, method =
"heartbeat")
+ int heartbeat(
+ @Param("id") long id, @Param("lastHeartbeat") long lastHeartbeat,
@Param("now") long now);
+
+ @SelectProvider(
+ type = IcebergCleanupJobSQLProviderFactory.class,
+ method = "selectUnfinishedJobId")
+ Long selectUnfinishedJobId(
+ @Param("catalogId") long catalogId,
+ @Param("namespace") String namespace,
+ @Param("table") String table);
+
+ @DeleteProvider(
+ type = IcebergCleanupJobSQLProviderFactory.class,
+ method = "deleteFinishedJobsByLegacyTimeline")
+ int deleteFinishedJobsByLegacyTimeline(@Param("legacyTimeline") long
legacyTimeline);
+
+ @SelectProvider(type = IcebergCleanupJobSQLProviderFactory.class, method =
"selectState")
+ String selectState(@Param("id") long id);
+}
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/mapper/IcebergCleanupJobSQLProviderFactory.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/mapper/IcebergCleanupJobSQLProviderFactory.java
new file mode 100644
index 0000000000..7c1969fc3f
--- /dev/null
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/mapper/IcebergCleanupJobSQLProviderFactory.java
@@ -0,0 +1,110 @@
+/*
+ * 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.cleanup.mapper;
+
+import com.google.common.collect.ImmutableMap;
+import java.util.Map;
+import
org.apache.gravitino.iceberg.service.cleanup.mapper.provider.base.IcebergCleanupJobBaseSQLProvider;
+import org.apache.gravitino.iceberg.service.cleanup.po.IcebergCleanupJobPO;
+import org.apache.gravitino.storage.relational.JDBCBackend.JDBCBackendType;
+import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
+import org.apache.ibatis.annotations.Param;
+
+/**
+ * Supplies the SQL for {@link IcebergCleanupJobMapper}, dispatching by the
entity store's
+ * configured JDBC backend. The statements are written with portable,
parameterized SQL (the row id
+ * and all timestamps are generated by the application, not the database), so
the {@link
+ * IcebergCleanupJobBaseSQLProvider} serves MySQL, H2, and PostgreSQL alike. A
backend that ever
+ * needs a divergent statement can be registered here by mapping its {@link
JDBCBackendType} to an
+ * {@link IcebergCleanupJobBaseSQLProvider} subclass that overrides only the
affected methods.
+ */
+public class IcebergCleanupJobSQLProviderFactory {
+
+ private static final IcebergCleanupJobBaseSQLProvider BASE_PROVIDER =
+ new IcebergCleanupJobBaseSQLProvider();
+
+ private static final Map<JDBCBackendType, IcebergCleanupJobBaseSQLProvider>
PROVIDERS =
+ ImmutableMap.of(
+ JDBCBackendType.MYSQL, BASE_PROVIDER,
+ JDBCBackendType.H2, BASE_PROVIDER,
+ JDBCBackendType.POSTGRESQL, BASE_PROVIDER);
+
+ private static IcebergCleanupJobBaseSQLProvider getProvider() {
+ String databaseId =
+ SqlSessionFactoryHelper.getInstance()
+ .getSqlSessionFactory()
+ .getConfiguration()
+ .getDatabaseId();
+ return PROVIDERS.get(JDBCBackendType.fromString(databaseId));
+ }
+
+ public static String insertCleanupJob(@Param("po") IcebergCleanupJobPO po) {
+ return getProvider().insertCleanupJob(po);
+ }
+
+ public static String selectCandidateJobs(
+ @Param("heartbeatExpiry") long heartbeatExpiry, @Param("window") int
window) {
+ return getProvider().selectCandidateJobs(heartbeatExpiry, window);
+ }
+
+ public static String markRunning(
+ @Param("id") long id,
+ @Param("now") long now,
+ @Param("heartbeatExpiry") long heartbeatExpiry) {
+ return getProvider().markRunning(id, now, heartbeatExpiry);
+ }
+
+ public static String markFinished(
+ @Param("id") long id,
+ @Param("state") String state,
+ @Param("reason") String reason,
+ @Param("now") long now) {
+ return getProvider().markFinished(id, state, reason, now);
+ }
+
+ public static String recordFailure(
+ @Param("id") long id,
+ @Param("reason") String reason,
+ @Param("maxAttempts") int maxAttempts,
+ @Param("now") long now) {
+ return getProvider().recordFailure(id, reason, maxAttempts, now);
+ }
+
+ public static String heartbeat(
+ @Param("id") long id, @Param("lastHeartbeat") long lastHeartbeat,
@Param("now") long now) {
+ return getProvider().heartbeat(id, lastHeartbeat, now);
+ }
+
+ public static String selectUnfinishedJobId(
+ @Param("catalogId") long catalogId,
+ @Param("namespace") String namespace,
+ @Param("table") String table) {
+ return getProvider().selectUnfinishedJobId(catalogId, namespace, table);
+ }
+
+ public static String deleteFinishedJobsByLegacyTimeline(
+ @Param("legacyTimeline") long legacyTimeline) {
+ return getProvider().deleteFinishedJobsByLegacyTimeline(legacyTimeline);
+ }
+
+ public static String selectState(@Param("id") long id) {
+ return getProvider().selectState(id);
+ }
+}
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
new file mode 100644
index 0000000000..cdfd225129
--- /dev/null
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/mapper/provider/IcebergCleanupMapperPackageProvider.java
@@ -0,0 +1,38 @@
+/*
+ * 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.cleanup.mapper.provider;
+
+import com.google.common.collect.ImmutableList;
+import java.util.List;
+import
org.apache.gravitino.iceberg.service.cleanup.mapper.IcebergCleanupJobMapper;
+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.
+ */
+public class IcebergCleanupMapperPackageProvider implements
MapperPackageProvider {
+
+ @Override
+ public List<Class<?>> getMapperClasses() {
+ return ImmutableList.of(IcebergCleanupJobMapper.class);
+ }
+}
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/mapper/provider/base/IcebergCleanupJobBaseSQLProvider.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/mapper/provider/base/IcebergCleanupJobBaseSQLProvider.java
new file mode 100644
index 0000000000..0de925bdf6
--- /dev/null
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/mapper/provider/base/IcebergCleanupJobBaseSQLProvider.java
@@ -0,0 +1,185 @@
+/*
+ * 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.cleanup.mapper.provider.base;
+
+import static
org.apache.gravitino.iceberg.service.cleanup.mapper.IcebergCleanupJobMapper.TABLE_NAME;
+
+import org.apache.gravitino.iceberg.service.cleanup.po.IcebergCleanupJobPO;
+import org.apache.ibatis.annotations.Param;
+
+/**
+ * Portable SQL for {@code iceberg_cleanup_job}, shared by all supported JDBC
backends. The
+ * statements use parameterized SQL with the row id and all timestamps
generated by the application
+ * (not the database), so this base provider serves MySQL, H2, and PostgreSQL
alike. A backend that
+ * ever needs a divergent statement can subclass this and override only the
affected methods.
+ */
+public class IcebergCleanupJobBaseSQLProvider {
+
+ /**
+ * @param po the job row to insert
+ * @return the INSERT statement
+ */
+ public String insertCleanupJob(@Param("po") IcebergCleanupJobPO po) {
+ return "INSERT INTO "
+ + TABLE_NAME
+ + " (id, catalog_id, namespace, table_name, metadata_location,"
+ + " file_io_impl, file_io_props, state, attempts, last_error,
heartbeat_at, created_by,"
+ + " updated_at) VALUES (#{po.id}, #{po.catalogId}, #{po.namespace},"
+ + " #{po.tableName}, #{po.metadataLocation}, #{po.fileIOImpl},
#{po.fileIOProps},"
+ + " #{po.state}, #{po.attempts}, #{po.lastError}, #{po.heartbeatAt},
#{po.createdBy},"
+ + " #{po.updatedAt})";
+ }
+
+ /**
+ * @param heartbeatExpiry heartbeats older than this make a RUNNING row
reclaimable
+ * @param window max candidate rows to return
+ * @return the candidate-window SELECT statement
+ */
+ public String selectCandidateJobs(
+ @Param("heartbeatExpiry") long heartbeatExpiry, @Param("window") int
window) {
+ // A candidate row is PENDING, or RUNNING whose worker has gone silent.
heartbeat_at is
+ // NOT NULL (it defaults to 0 and markRunning always writes the current
time), so the
+ // staleness test is a plain comparison with no NULL branch and no
three-valued-logic
+ // trap, which also keeps the predicate index-friendly. Full rows are
returned so a winning
+ // claimer can build the job directly: the columns the job needs are
immutable after enqueue,
+ // so the pre-claim snapshot stays accurate without a re-read.
+ return "SELECT id, catalog_id AS catalogId, namespace,"
+ + " table_name AS tableName, metadata_location AS metadataLocation,"
+ + " file_io_impl AS fileIOImpl, file_io_props AS fileIOProps, state,
attempts,"
+ + " last_error AS lastError, heartbeat_at AS heartbeatAt, created_by
AS createdBy,"
+ + " updated_at AS updatedAt FROM "
+ + TABLE_NAME
+ + " WHERE state = 'PENDING'"
+ + " OR (state = 'RUNNING' AND heartbeat_at < #{heartbeatExpiry})"
+ + " ORDER BY updated_at LIMIT #{window}";
+ }
+
+ /**
+ * @param id job id
+ * @param now claim timestamp, written as state change and initial heartbeat
+ * @param heartbeatExpiry heartbeats older than this make a RUNNING row
reclaimable
+ * @return the claim UPDATE statement
+ */
+ public String markRunning(
+ @Param("id") long id,
+ @Param("now") long now,
+ @Param("heartbeatExpiry") long heartbeatExpiry) {
+ return "UPDATE "
+ + TABLE_NAME
+ + " SET state = 'RUNNING', heartbeat_at = #{now}, updated_at = #{now}"
+ + " WHERE id = #{id} AND (state = 'PENDING'"
+ + " OR (state = 'RUNNING' AND heartbeat_at < #{heartbeatExpiry}))";
+ }
+
+ /**
+ * @param id job id
+ * @param state terminal state to set (SUCCEEDED or FAILED)
+ * @param reason failure reason, or {@code null} for success
+ * @param now state-change timestamp
+ * @return the terminal-transition UPDATE statement
+ */
+ public String markFinished(
+ @Param("id") long id,
+ @Param("state") String state,
+ @Param("reason") String reason,
+ @Param("now") long now) {
+ // Shared transition to a final state: SUCCEEDED (reason null) or FAILED
(reason set).
+ return "UPDATE "
+ + TABLE_NAME
+ + " SET state = #{state}, last_error = #{reason}, heartbeat_at = 0,"
+ + " updated_at = #{now} WHERE id = #{id} AND state = 'RUNNING'";
+ }
+
+ /**
+ * @param id job id
+ * @param reason failure reason
+ * @param maxAttempts attempt ceiling past which the job is FAILED
+ * @param now state-change timestamp
+ * @return the transient-failure UPDATE statement
+ */
+ public String recordFailure(
+ @Param("id") long id,
+ @Param("reason") String reason,
+ @Param("maxAttempts") int maxAttempts,
+ @Param("now") long now) {
+ // The state CASE must observe the pre-increment attempts value, so it
MUST precede
+ // "attempts = attempts + 1" in the SET list. MySQL evaluates SET
assignments left to
+ // right and later clauses see already-updated columns; if the increment
came first the
+ // CASE would compare attempts + 2 and fail one attempt early. H2 and
PostgreSQL evaluate
+ // every right-hand side against the original row, so state-first is
correct there too,
+ // keeping this statement portable across all backends.
+ return "UPDATE "
+ + TABLE_NAME
+ + " SET state = CASE WHEN attempts + 1 >= #{maxAttempts} THEN 'FAILED'
ELSE 'PENDING' END,"
+ + " attempts = attempts + 1, last_error = #{reason},"
+ + " heartbeat_at = 0, updated_at = #{now} WHERE id = #{id} AND state =
'RUNNING'";
+ }
+
+ /**
+ * @param id job id
+ * @param lastHeartbeat previous heartbeat value the caller wrote
(compare-and-swap key)
+ * @param now new heartbeat value
+ * @return the heartbeat UPDATE statement
+ */
+ public String heartbeat(
+ @Param("id") long id, @Param("lastHeartbeat") long lastHeartbeat,
@Param("now") long now) {
+ return "UPDATE "
+ + TABLE_NAME
+ + " SET heartbeat_at = #{now}, updated_at = #{now}"
+ + " WHERE id = #{id} AND state = 'RUNNING' AND heartbeat_at =
#{lastHeartbeat}";
+ }
+
+ /**
+ * @param catalogId globally unique id of the owning catalog
+ * @param namespace table namespace
+ * @param table table name
+ * @return the unfinished-job lookup SELECT statement
+ */
+ public String selectUnfinishedJobId(
+ @Param("catalogId") long catalogId,
+ @Param("namespace") String namespace,
+ @Param("table") String table) {
+ // catalog_id is globally unique (catalog_meta's primary key), so it
identifies the catalog
+ // without scoping by metalake, and it is stable across catalog rename
unlike a name.
+ return "SELECT id FROM "
+ + TABLE_NAME
+ + " WHERE catalog_id = #{catalogId}"
+ + " AND namespace = #{namespace} AND table_name = #{table}"
+ + " AND state IN ('PENDING', 'RUNNING') LIMIT 1";
+ }
+
+ /**
+ * @param legacyTimeline cutoff; finished rows updated before this are
deleted
+ * @return the cleanup DELETE statement
+ */
+ public String deleteFinishedJobsByLegacyTimeline(@Param("legacyTimeline")
long legacyTimeline) {
+ return "DELETE FROM "
+ + TABLE_NAME
+ + " WHERE state IN ('SUCCEEDED', 'FAILED') AND updated_at <
#{legacyTimeline}";
+ }
+
+ /**
+ * @param id job id
+ * @return the state SELECT statement
+ */
+ public String selectState(@Param("id") long id) {
+ return "SELECT state FROM " + TABLE_NAME + " WHERE id = #{id}";
+ }
+}
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/po/IcebergCleanupJobPO.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/po/IcebergCleanupJobPO.java
new file mode 100644
index 0000000000..d9faa298b9
--- /dev/null
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/po/IcebergCleanupJobPO.java
@@ -0,0 +1,122 @@
+/*
+ * 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.cleanup.po;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.Map;
+import lombok.AccessLevel;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import org.apache.gravitino.iceberg.service.cleanup.IcebergCleanupJob;
+import
org.apache.gravitino.iceberg.service.cleanup.mapper.IcebergCleanupJobMapper;
+import org.apache.gravitino.json.JsonUtils;
+
+/**
+ * Persistence object for one {@code iceberg_cleanup_job} row. Mirrors the
table columns and is
+ * mapped by {@link IcebergCleanupJobMapper}, and converts to and from the
{@link IcebergCleanupJob}
+ * domain object via {@link #fromCleanupJob} and {@link #toCleanupJob}.
Instances are created
+ * through the generated {@code builder()} (or populated by MyBatis via field
reflection through the
+ * private no-arg constructor) and expose no setters, so they are effectively
immutable once built.
+ */
+@Getter
+@Builder(setterPrefix = "with")
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+@AllArgsConstructor(access = AccessLevel.PRIVATE)
+public class IcebergCleanupJobPO {
+
+ private Long id;
+ private Long catalogId;
+ private String namespace;
+ private String tableName;
+ private String metadataLocation;
+ private String fileIOImpl;
+ private String fileIOProps;
+ private String state;
+ private Integer attempts;
+ private String lastError;
+ private Long heartbeatAt;
+ private String createdBy;
+ private Long updatedAt;
+
+ /**
+ * Builds a new PENDING row from a domain job.
+ *
+ * @param job the domain job to persist
+ * @param id the generated row id
+ * @param now the enqueue timestamp (initial {@code updated_at})
+ * @return the persistence object
+ */
+ public static IcebergCleanupJobPO fromCleanupJob(IcebergCleanupJob job, long
id, long now) {
+ return builder()
+ .withId(id)
+ .withCatalogId(job.catalogId())
+ .withNamespace(job.namespace())
+ .withTableName(job.tableName())
+ .withMetadataLocation(job.metadataLocation())
+ .withFileIOImpl(job.fileIOImpl())
+ .withFileIOProps(propertiesToJson(job.fileIOProperties()))
+ .withState(IcebergCleanupJob.State.PENDING.name())
+ .withAttempts(0)
+ .withLastError(null)
+ .withHeartbeatAt(0L)
+ .withCreatedBy(job.createdBy())
+ .withUpdatedAt(now)
+ .build();
+ }
+
+ /**
+ * Converts this row to the domain job. Only the columns that are immutable
after enqueue are
+ * carried over; mutable bookkeeping columns (state, attempts, heartbeat,
etc.) live only in the
+ * row.
+ *
+ * @return the domain job
+ */
+ public IcebergCleanupJob toCleanupJob() {
+ return new IcebergCleanupJob(
+ id,
+ catalogId,
+ namespace,
+ tableName,
+ metadataLocation,
+ fileIOImpl,
+ jsonToProperties(fileIOProps),
+ createdBy);
+ }
+
+ private static String propertiesToJson(Map<String, String> props) {
+ try {
+ return JsonUtils.objectMapper().writeValueAsString(props);
+ } catch (IOException e) {
+ throw new UncheckedIOException("Failed to serialize fileIOProperties",
e);
+ }
+ }
+
+ private static Map<String, String> jsonToProperties(String json) {
+ try {
+ return JsonUtils.objectMapper().readValue(json, new
TypeReference<Map<String, String>>() {});
+ } catch (IOException e) {
+ throw new UncheckedIOException("Failed to deserialize fileIOProperties",
e);
+ }
+ }
+}
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
new file mode 100644
index 0000000000..7ed1ef96c3
--- /dev/null
+++
b/iceberg/iceberg-rest-server/src/main/resources/META-INF/services/org.apache.gravitino.storage.relational.mapper.provider.MapperPackageProvider
@@ -0,0 +1,19 @@
+#
+# 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/service/cleanup/AbstractIcebergCleanupJobStoreBackendTest.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/cleanup/AbstractIcebergCleanupJobStoreBackendTest.java
new file mode 100644
index 0000000000..3e58e925df
--- /dev/null
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/cleanup/AbstractIcebergCleanupJobStoreBackendTest.java
@@ -0,0 +1,146 @@
+/*
+ * 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.cleanup;
+
+import com.google.common.collect.ImmutableMap;
+import java.util.Optional;
+import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.gravitino.GravitinoEnv;
+import org.apache.gravitino.storage.RandomIdGenerator;
+import org.apache.gravitino.storage.relational.TestJDBCBackend;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+
+/**
+ * Shared cleanup-store test logic exercised against the same relational
backend matrix as core
+ * metadata service tests. {@link TestJDBCBackend} initializes H2 by default,
adds MySQL and
+ * PostgreSQL when {@code dockerTest=true}, and truncates all backend tables
before each invocation.
+ */
+abstract class AbstractIcebergCleanupJobStoreBackendTest extends
TestJDBCBackend {
+
+ private IcebergCleanupJobStore store;
+
+ // TestJDBCBackend's BackendTestExtension overwrites GravitinoEnv's
singleton "config" and
+ // "idGenerator" fields with a backend-only Mockito mock and never restores
them. Because the
+ // whole iceberg-rest-server module runs in one JVM, that mock would leak
into later test classes
+ // (e.g. credential vending), where
MetadataAuthzHelper.enableAuthorization() unboxes the
+ // unstubbed config.get(ENABLE_AUTHORIZATION) -> null and NPEs. Snapshot the
pre-test fields and
+ // restore them after this class so it leaves GravitinoEnv exactly as it
found it.
+ private Object originalConfig;
+ private Object originalIdGenerator;
+
+ @BeforeAll
+ public void snapshotGravitinoEnv() throws IllegalAccessException {
+ originalConfig = FieldUtils.readField(GravitinoEnv.getInstance(),
"config", true);
+ originalIdGenerator = FieldUtils.readField(GravitinoEnv.getInstance(),
"idGenerator", true);
+ }
+
+ @AfterAll
+ public void restoreGravitinoEnv() throws IllegalAccessException {
+ FieldUtils.writeField(GravitinoEnv.getInstance(), "config",
originalConfig, true);
+ FieldUtils.writeField(GravitinoEnv.getInstance(), "idGenerator",
originalIdGenerator, true);
+ }
+
+ @BeforeEach
+ public void prepareCleanupJobStore() {
+ store = new IcebergCleanupJobStore(new RandomIdGenerator());
+ }
+
+ private static final long CATALOG_ID = 100L;
+ private static final long OTHER_CATALOG_ID = 200L;
+
+ private static IcebergCleanupJob sampleJob() {
+ return new IcebergCleanupJob(
+ 0L,
+ CATALOG_ID,
+ "db",
+ "t",
+ "s3://b/db/t/metadata/0.json",
+ "org.apache.iceberg.aws.s3.S3FileIO",
+ ImmutableMap.of("k", "v"),
+ "alice");
+ }
+
+ @TestTemplate
+ void testAddTakeSucceedLifecycle() {
+ Assertions.assertFalse(store.findUnfinishedJobId(CATALOG_ID, "db",
"t").isPresent());
+
+ long id = store.addJob(sampleJob());
+ Assertions.assertTrue(id > 0);
+ Assertions.assertTrue(store.findUnfinishedJobId(CATALOG_ID, "db",
"t").isPresent());
+ // The same namespace/table under a different catalog must not match.
+ Assertions.assertFalse(store.findUnfinishedJobId(OTHER_CATALOG_ID, "db",
"t").isPresent());
+
+ long now = System.currentTimeMillis();
+ Optional<IcebergCleanupJob> taken = store.takePendingJob(now, 300_000L,
10);
+ Assertions.assertTrue(taken.isPresent());
+ Assertions.assertEquals(id, taken.get().id());
+ Assertions.assertEquals(ImmutableMap.of("k", "v"),
taken.get().fileIOProperties());
+ Assertions.assertEquals(IcebergCleanupJob.State.RUNNING,
store.stateOf(id));
+ Assertions.assertTrue(store.findUnfinishedJobId(CATALOG_ID, "db",
"t").isPresent());
+ Assertions.assertFalse(store.takePendingJob(now, 300_000L,
10).isPresent());
+
+ Assertions.assertTrue(store.markSucceeded(id));
+ Assertions.assertEquals(IcebergCleanupJob.State.SUCCEEDED,
store.stateOf(id));
+ // A second transition no longer owns the (now terminal) row, so it
reports no update.
+ Assertions.assertFalse(store.markSucceeded(id));
+ Assertions.assertFalse(store.findUnfinishedJobId(CATALOG_ID, "db",
"t").isPresent());
+ Assertions.assertEquals(
+ 1, store.deleteFinishedJobsByLegacyTimeline(System.currentTimeMillis()
+ 1));
+ }
+
+ @TestTemplate
+ void testMarkFailed() {
+ long id = store.addJob(sampleJob());
+ store.takePendingJob(System.currentTimeMillis(), 300_000L, 10);
+ Assertions.assertTrue(store.markFailed(id, "corrupt metadata"));
+ Assertions.assertEquals(IcebergCleanupJob.State.FAILED, store.stateOf(id));
+ }
+
+ @TestTemplate
+ void testTransientFailureRetriesThenFailsAtCeiling() {
+ long id = store.addJob(sampleJob());
+ for (int i = 0; i < 2; i++) {
+ store.takePendingJob(System.currentTimeMillis(), 300_000L, 10);
+ Assertions.assertTrue(store.recordFailure(id, "boom " + i, 3));
+ Assertions.assertEquals(IcebergCleanupJob.State.PENDING,
store.stateOf(id));
+ }
+ store.takePendingJob(System.currentTimeMillis(), 300_000L, 10);
+ Assertions.assertTrue(store.recordFailure(id, "boom final", 3));
+ Assertions.assertEquals(IcebergCleanupJob.State.FAILED, store.stateOf(id));
+ }
+
+ @TestTemplate
+ void testHeartbeatCasAndStaleTakeover() {
+ long id = store.addJob(sampleJob());
+ long t0 = System.currentTimeMillis();
+ store.takePendingJob(t0, 300_000L, 10);
+ Assertions.assertTrue(store.heartbeat(id, t0, t0 + 1000));
+ Assertions.assertFalse(store.heartbeat(id, t0, t0 + 2000));
+ // A stale RUNNING job can be taken again once its heartbeat ages past the
timeout.
+ Assertions.assertEquals(
+ id, store.takePendingJob(t0 + 400_000L, 300_000L,
10).orElseThrow().id());
+ }
+}
+
+class TestIcebergCleanupJobStoreBackend extends
AbstractIcebergCleanupJobStoreBackendTest {}
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/cleanup/TestIcebergCleanupJob.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/cleanup/TestIcebergCleanupJob.java
new file mode 100644
index 0000000000..260a3711c4
--- /dev/null
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/cleanup/TestIcebergCleanupJob.java
@@ -0,0 +1,53 @@
+/*
+ * 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.cleanup;
+
+import com.google.common.collect.ImmutableMap;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class TestIcebergCleanupJob {
+
+ @Test
+ void testConstructorAndAccessors() {
+ IcebergCleanupJob job =
+ new IcebergCleanupJob(
+ 0L,
+ 42L,
+ "db",
+ "t",
+ "s3://b/db/t/metadata/0.json",
+ "org.apache.iceberg.aws.s3.S3FileIO",
+ ImmutableMap.of("k", "v"),
+ "alice");
+ Assertions.assertEquals(42L, job.catalogId());
+ Assertions.assertEquals("db", job.namespace());
+ Assertions.assertEquals(ImmutableMap.of("k", "v"), job.fileIOProperties());
+ Assertions.assertEquals("alice", job.createdBy());
+ }
+
+ @Test
+ void testFinishedStates() {
+ Assertions.assertTrue(IcebergCleanupJob.State.SUCCEEDED.isFinished());
+ Assertions.assertTrue(IcebergCleanupJob.State.FAILED.isFinished());
+ Assertions.assertFalse(IcebergCleanupJob.State.PENDING.isFinished());
+ Assertions.assertFalse(IcebergCleanupJob.State.RUNNING.isFinished());
+ }
+}
diff --git
a/integration-test-common/src/test/java/org/apache/gravitino/integration/test/util/TestDatabaseName.java
b/integration-test-common/src/test/java/org/apache/gravitino/integration/test/util/TestDatabaseName.java
index 1c178036e2..39b644eb65 100644
---
a/integration-test-common/src/test/java/org/apache/gravitino/integration/test/util/TestDatabaseName.java
+++
b/integration-test-common/src/test/java/org/apache/gravitino/integration/test/util/TestDatabaseName.java
@@ -119,6 +119,8 @@ public enum TestDatabaseName {
},
PG_ICEBERG_AUTHZ_IT,
+ PG_ICEBERG_ASYNC_CLEANUP_IT,
+
CLICKHOUSE_CLICKHOUSE_ABSTRACT_IT,
CLICKHOUSE_CATALOG_CLICKHOUSE_IT,
CLICKHOUSE_AUDIT_CATALOG_CLICKHOUSE_IT,
diff --git a/scripts/h2/schema-1.3.0-h2.sql b/scripts/h2/schema-1.3.0-h2.sql
index fc6f026ec3..17de756468 100644
--- a/scripts/h2/schema-1.3.0-h2.sql
+++ b/scripts/h2/schema-1.3.0-h2.sql
@@ -612,3 +612,22 @@ CREATE TABLE IF NOT EXISTS `entity_change_log` (
PRIMARY KEY (`id`),
KEY `idx_ecl_created_at` (`created_at`)
) ENGINE=InnoDB COMMENT='Append-only log of entity structural changes for
targeted metadataIdCache invalidation';
+
+CREATE TABLE IF NOT EXISTS `iceberg_cleanup_job` (
+ `id` BIGINT NOT NULL COMMENT 'globally unique cleanup
job id',
+ `catalog_id` BIGINT NOT NULL COMMENT 'globally unique id of
the owning catalog, stable across catalog rename',
+ `namespace` VARCHAR(512) NOT NULL COMMENT 'namespace of the table
to be cleaned up',
+ `table_name` VARCHAR(256) NOT NULL COMMENT 'name of the table to be
cleaned up',
+ `metadata_location` CLOB NOT NULL COMMENT 'location of the table
metadata file to purge',
+ `file_io_impl` VARCHAR(256) NOT NULL COMMENT 'FileIO implementation
class used to access the table files',
+ `file_io_props` CLOB NOT NULL COMMENT 'JSON-encoded FileIO
properties',
+ `state` VARCHAR(16) NOT NULL COMMENT
'PENDING|RUNNING|SUCCEEDED|FAILED',
+ `attempts` INT NOT NULL DEFAULT 0 COMMENT 'number of
processing attempts made so far',
+ `last_error` VARCHAR(2048) NULL COMMENT 'truncated reason for the
most recent failure, NULL until a job fails',
+ `heartbeat_at` BIGINT NOT NULL DEFAULT 0 COMMENT 'last heartbeat
from the worker, 0 when not running',
+ `created_by` VARCHAR(128) NOT NULL COMMENT 'principal that requested
the drop (audit)',
+ `updated_at` BIGINT NOT NULL COMMENT 'last state change,
drives poll ordering and old finished-job cleanup',
+ PRIMARY KEY (`id`)
+) COMMENT='async Iceberg table cleanup jobs';
+CREATE INDEX IF NOT EXISTS `idx_state_updated` ON `iceberg_cleanup_job`
(`state`, `updated_at`);
+CREATE INDEX IF NOT EXISTS `idx_object` ON `iceberg_cleanup_job`
(`catalog_id`, `namespace`, `table_name`, `state`);
diff --git a/scripts/h2/upgrade-1.2.0-to-1.3.0-h2.sql
b/scripts/h2/upgrade-1.2.0-to-1.3.0-h2.sql
index 5366a16436..adc350b6d6 100644
--- a/scripts/h2/upgrade-1.2.0-to-1.3.0-h2.sql
+++ b/scripts/h2/upgrade-1.2.0-to-1.3.0-h2.sql
@@ -100,3 +100,22 @@ CREATE TABLE IF NOT EXISTS `idp_user_group_rel` (
KEY `idx_iuig_uid` (`user_id`),
KEY `idx_iuig_gid` (`group_id`)
) ENGINE=InnoDB;
+
+CREATE TABLE IF NOT EXISTS `iceberg_cleanup_job` (
+ `id` BIGINT NOT NULL COMMENT 'globally unique cleanup
job id',
+ `catalog_id` BIGINT NOT NULL COMMENT 'globally unique id of
the owning catalog, stable across catalog rename',
+ `namespace` VARCHAR(512) NOT NULL COMMENT 'namespace of the table
to be cleaned up',
+ `table_name` VARCHAR(256) NOT NULL COMMENT 'name of the table to be
cleaned up',
+ `metadata_location` CLOB NOT NULL COMMENT 'location of the table
metadata file to purge',
+ `file_io_impl` VARCHAR(256) NOT NULL COMMENT 'FileIO implementation
class used to access the table files',
+ `file_io_props` CLOB NOT NULL COMMENT 'JSON-encoded FileIO
properties',
+ `state` VARCHAR(16) NOT NULL COMMENT
'PENDING|RUNNING|SUCCEEDED|FAILED',
+ `attempts` INT NOT NULL DEFAULT 0 COMMENT 'number of
processing attempts made so far',
+ `last_error` VARCHAR(2048) NULL COMMENT 'truncated reason for the
most recent failure, NULL until a job fails',
+ `heartbeat_at` BIGINT NOT NULL DEFAULT 0 COMMENT 'last heartbeat
from the worker, 0 when not running',
+ `created_by` VARCHAR(128) NOT NULL COMMENT 'principal that requested
the drop (audit)',
+ `updated_at` BIGINT NOT NULL COMMENT 'last state change,
drives poll ordering and old finished-job cleanup',
+ PRIMARY KEY (`id`)
+) COMMENT='async Iceberg table cleanup jobs';
+CREATE INDEX IF NOT EXISTS `idx_state_updated` ON `iceberg_cleanup_job`
(`state`, `updated_at`);
+CREATE INDEX IF NOT EXISTS `idx_object` ON `iceberg_cleanup_job`
(`catalog_id`, `namespace`, `table_name`, `state`);
diff --git a/scripts/mysql/schema-1.3.0-mysql.sql
b/scripts/mysql/schema-1.3.0-mysql.sql
index 7ed2d03e17..5959b1886d 100644
--- a/scripts/mysql/schema-1.3.0-mysql.sql
+++ b/scripts/mysql/schema-1.3.0-mysql.sql
@@ -599,3 +599,22 @@ CREATE TABLE IF NOT EXISTS `entity_change_log` (
PRIMARY KEY (`id`),
KEY `idx_ecl_created_at` (`created_at`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT
'Append-only log of entity structural changes for targeted metadataIdCache
invalidation';
+
+CREATE TABLE IF NOT EXISTS `iceberg_cleanup_job` (
+ `id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'globally unique
cleanup job id',
+ `catalog_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'globally unique
id of the owning catalog, stable across catalog rename',
+ `namespace` VARCHAR(512) NOT NULL COMMENT 'namespace of the table
to be cleaned up',
+ `table_name` VARCHAR(256) NOT NULL COMMENT 'name of the table to be
cleaned up',
+ `metadata_location` MEDIUMTEXT NOT NULL COMMENT 'location of the table
metadata file to purge',
+ `file_io_impl` VARCHAR(256) NOT NULL COMMENT 'FileIO implementation
class used to access the table files',
+ `file_io_props` MEDIUMTEXT NOT NULL COMMENT 'JSON-encoded FileIO
properties',
+ `state` VARCHAR(16) NOT NULL COMMENT
'PENDING|RUNNING|SUCCEEDED|FAILED',
+ `attempts` INT(10) NOT NULL DEFAULT 0 COMMENT 'number of
processing attempts made so far',
+ `last_error` VARCHAR(2048) NULL COMMENT 'truncated reason for the
most recent failure, NULL until a job fails',
+ `heartbeat_at` BIGINT(20) NOT NULL DEFAULT 0 COMMENT 'last heartbeat
from the worker, 0 when not running',
+ `created_by` VARCHAR(128) NOT NULL COMMENT 'principal that requested
the drop (audit)',
+ `updated_at` BIGINT(20) NOT NULL COMMENT 'last state change,
drives poll ordering and old finished-job cleanup',
+ PRIMARY KEY (`id`),
+ KEY `idx_state_updated` (`state`, `updated_at`),
+ KEY `idx_object` (`catalog_id`, `namespace`(255), `table_name`(128), `state`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT 'async
Iceberg table cleanup jobs';
diff --git a/scripts/mysql/upgrade-1.2.0-to-1.3.0-mysql.sql
b/scripts/mysql/upgrade-1.2.0-to-1.3.0-mysql.sql
index d31b6314b0..4c4ef09ce5 100644
--- a/scripts/mysql/upgrade-1.2.0-to-1.3.0-mysql.sql
+++ b/scripts/mysql/upgrade-1.2.0-to-1.3.0-mysql.sql
@@ -118,3 +118,22 @@ CREATE TABLE IF NOT EXISTS `idp_user_group_rel` (
KEY `idx_iuig_uid` (`user_id`),
KEY `idx_iuig_gid` (`group_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT 'local IdP
user group relation';
+
+CREATE TABLE IF NOT EXISTS `iceberg_cleanup_job` (
+ `id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'globally unique
cleanup job id',
+ `catalog_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'globally unique
id of the owning catalog, stable across catalog rename',
+ `namespace` VARCHAR(512) NOT NULL COMMENT 'namespace of the table
to be cleaned up',
+ `table_name` VARCHAR(256) NOT NULL COMMENT 'name of the table to be
cleaned up',
+ `metadata_location` MEDIUMTEXT NOT NULL COMMENT 'location of the table
metadata file to purge',
+ `file_io_impl` VARCHAR(256) NOT NULL COMMENT 'FileIO implementation
class used to access the table files',
+ `file_io_props` MEDIUMTEXT NOT NULL COMMENT 'JSON-encoded FileIO
properties',
+ `state` VARCHAR(16) NOT NULL COMMENT
'PENDING|RUNNING|SUCCEEDED|FAILED',
+ `attempts` INT(10) NOT NULL DEFAULT 0 COMMENT 'number of
processing attempts made so far',
+ `last_error` VARCHAR(2048) NULL COMMENT 'truncated reason for the
most recent failure, NULL until a job fails',
+ `heartbeat_at` BIGINT(20) NOT NULL DEFAULT 0 COMMENT 'last heartbeat
from the worker, 0 when not running',
+ `created_by` VARCHAR(128) NOT NULL COMMENT 'principal that requested
the drop (audit)',
+ `updated_at` BIGINT(20) NOT NULL COMMENT 'last state change,
drives poll ordering and old finished-job cleanup',
+ PRIMARY KEY (`id`),
+ KEY `idx_state_updated` (`state`, `updated_at`),
+ KEY `idx_object` (`catalog_id`, `namespace`(255), `table_name`(128), `state`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT 'async
Iceberg table cleanup jobs';
diff --git a/scripts/postgresql/schema-1.3.0-postgresql.sql
b/scripts/postgresql/schema-1.3.0-postgresql.sql
index b922b369da..8efd931d94 100644
--- a/scripts/postgresql/schema-1.3.0-postgresql.sql
+++ b/scripts/postgresql/schema-1.3.0-postgresql.sql
@@ -1054,3 +1054,35 @@ COMMENT ON COLUMN entity_change_log.entity_type IS
'METALAKE | CATALOG | SCHEMA
COMMENT ON COLUMN entity_change_log.entity_full_name IS 'Dot-separated full
name of the affected entity. For ALTER, stores the old name. For DROP, stores
the entity name.';
COMMENT ON COLUMN entity_change_log.operate_type IS 'Operate type code:
1=ALTER, 2=DROP, 3=INSERT. Codes are stable and never re-used.';
COMMENT ON COLUMN entity_change_log.created_at IS 'timestamp of the change in
millis';
+
+CREATE TABLE IF NOT EXISTS iceberg_cleanup_job (
+ id BIGINT NOT NULL PRIMARY KEY,
+ catalog_id BIGINT NOT NULL,
+ namespace VARCHAR(512) NOT NULL,
+ table_name VARCHAR(256) NOT NULL,
+ metadata_location TEXT NOT NULL,
+ file_io_impl VARCHAR(256) NOT NULL,
+ file_io_props TEXT NOT NULL,
+ state VARCHAR(16) NOT NULL,
+ attempts INT NOT NULL DEFAULT 0,
+ last_error VARCHAR(2048),
+ heartbeat_at BIGINT NOT NULL DEFAULT 0,
+ created_by VARCHAR(128) NOT NULL,
+ updated_at BIGINT NOT NULL
+);
+CREATE INDEX IF NOT EXISTS idx_state_updated ON iceberg_cleanup_job (state,
updated_at);
+CREATE INDEX IF NOT EXISTS idx_object ON iceberg_cleanup_job (catalog_id,
namespace, table_name, state);
+COMMENT ON TABLE iceberg_cleanup_job IS 'async Iceberg table cleanup jobs';
+COMMENT ON COLUMN iceberg_cleanup_job.id IS 'globally unique cleanup job id';
+COMMENT ON COLUMN iceberg_cleanup_job.catalog_id IS 'globally unique id of the
owning catalog, stable across catalog rename';
+COMMENT ON COLUMN iceberg_cleanup_job.namespace IS 'namespace of the table to
be cleaned up';
+COMMENT ON COLUMN iceberg_cleanup_job.table_name IS 'name of the table to be
cleaned up';
+COMMENT ON COLUMN iceberg_cleanup_job.metadata_location IS 'location of the
table metadata file to purge';
+COMMENT ON COLUMN iceberg_cleanup_job.file_io_impl IS 'FileIO implementation
class used to access the table files';
+COMMENT ON COLUMN iceberg_cleanup_job.file_io_props IS 'JSON-encoded FileIO
properties';
+COMMENT ON COLUMN iceberg_cleanup_job.state IS 'PENDING | RUNNING | SUCCEEDED
| FAILED';
+COMMENT ON COLUMN iceberg_cleanup_job.attempts IS 'number of processing
attempts made so far';
+COMMENT ON COLUMN iceberg_cleanup_job.last_error IS 'truncated reason for the
most recent failure, NULL until a job fails';
+COMMENT ON COLUMN iceberg_cleanup_job.heartbeat_at IS 'last heartbeat from the
worker, 0 when not running';
+COMMENT ON COLUMN iceberg_cleanup_job.created_by IS 'principal that requested
the drop (audit)';
+COMMENT ON COLUMN iceberg_cleanup_job.updated_at IS 'last state change, drives
poll ordering and old finished-job cleanup';
diff --git a/scripts/postgresql/upgrade-1.2.0-to-1.3.0-postgresql.sql
b/scripts/postgresql/upgrade-1.2.0-to-1.3.0-postgresql.sql
index bf3071567d..03c362cc88 100644
--- a/scripts/postgresql/upgrade-1.2.0-to-1.3.0-postgresql.sql
+++ b/scripts/postgresql/upgrade-1.2.0-to-1.3.0-postgresql.sql
@@ -151,3 +151,35 @@ COMMENT ON COLUMN idp_user_group_rel.group_id IS 'idp
group id';
COMMENT ON COLUMN idp_user_group_rel.current_version IS 'idp relation current
version';
COMMENT ON COLUMN idp_user_group_rel.last_version IS 'idp relation last
version';
COMMENT ON COLUMN idp_user_group_rel.deleted_at IS 'idp relation deleted at';
+
+CREATE TABLE IF NOT EXISTS iceberg_cleanup_job (
+ id BIGINT NOT NULL PRIMARY KEY,
+ catalog_id BIGINT NOT NULL,
+ namespace VARCHAR(512) NOT NULL,
+ table_name VARCHAR(256) NOT NULL,
+ metadata_location TEXT NOT NULL,
+ file_io_impl VARCHAR(256) NOT NULL,
+ file_io_props TEXT NOT NULL,
+ state VARCHAR(16) NOT NULL,
+ attempts INT NOT NULL DEFAULT 0,
+ last_error VARCHAR(2048),
+ heartbeat_at BIGINT NOT NULL DEFAULT 0,
+ created_by VARCHAR(128) NOT NULL,
+ updated_at BIGINT NOT NULL
+);
+CREATE INDEX IF NOT EXISTS idx_state_updated ON iceberg_cleanup_job (state,
updated_at);
+CREATE INDEX IF NOT EXISTS idx_object ON iceberg_cleanup_job (catalog_id,
namespace, table_name, state);
+COMMENT ON TABLE iceberg_cleanup_job IS 'async Iceberg table cleanup jobs';
+COMMENT ON COLUMN iceberg_cleanup_job.id IS 'globally unique cleanup job id';
+COMMENT ON COLUMN iceberg_cleanup_job.catalog_id IS 'globally unique id of the
owning catalog, stable across catalog rename';
+COMMENT ON COLUMN iceberg_cleanup_job.namespace IS 'namespace of the table to
be cleaned up';
+COMMENT ON COLUMN iceberg_cleanup_job.table_name IS 'name of the table to be
cleaned up';
+COMMENT ON COLUMN iceberg_cleanup_job.metadata_location IS 'location of the
table metadata file to purge';
+COMMENT ON COLUMN iceberg_cleanup_job.file_io_impl IS 'FileIO implementation
class used to access the table files';
+COMMENT ON COLUMN iceberg_cleanup_job.file_io_props IS 'JSON-encoded FileIO
properties';
+COMMENT ON COLUMN iceberg_cleanup_job.state IS 'PENDING | RUNNING | SUCCEEDED
| FAILED';
+COMMENT ON COLUMN iceberg_cleanup_job.attempts IS 'number of processing
attempts made so far';
+COMMENT ON COLUMN iceberg_cleanup_job.last_error IS 'truncated reason for the
most recent failure, NULL until a job fails';
+COMMENT ON COLUMN iceberg_cleanup_job.heartbeat_at IS 'last heartbeat from the
worker, 0 when not running';
+COMMENT ON COLUMN iceberg_cleanup_job.created_by IS 'principal that requested
the drop (audit)';
+COMMENT ON COLUMN iceberg_cleanup_job.updated_at IS 'last state change, drives
poll ordering and old finished-job cleanup';
