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 4f11ac17b8 [#13761] feat(tms): plugins/tms skeleton, DDL, and DAOs
(#13762)
4f11ac17b8 is described below
commit 4f11ac17b84134a766fcf3ed0396ac7a04302d11
Author: MaSai <[email protected]>
AuthorDate: Tue Oct 13 09:55:44 2026 +0800
[#13761] feat(tms): plugins/tms skeleton, DDL, and DAOs (#13762)
### What changes were proposed in this pull request?
- Add `plugins/tms` module (same packaging pattern as `idp-basic`, not
in `core`)
- DDL for `table_maintenance_action` and `table_snapshot_metrics` (MySQL
/ H2 / PostgreSQL), including 1.3.0→2.0.0 upgrade scripts
- Relational DAOs (POs, MyBatis mappers / SQL providers, meta services)
+ `TmsMapperPackageProvider` SPI
- Action lifecycle: insert in-flight occupancy → bind optional `job_id`
after `runJob`; S0 / finish updates are one-shot; metrics insert is
idempotent and not deleted with the action
- Distribution `copyLibAndConfigs` and DAO unit tests
**Out of scope:** db-scheduler task tables / pools, REST Feature
lifecycle, CAS / occupancy gates, and in-job sample / drift / recommend
(follow-up subtasks under #13760).
### Why are the changes needed?
First subtask of Epic #13760: TMS persistence foundation for the
in-process Table Maintenance Service.
Fix: #13761
### Does this PR introduce _any_ user-facing change?
New relational tables on fresh 2.0.0 installs. Plugin jar is optional
until enabled by later Feature / scheduler work.
### How was this patch tested?
```bash
./gradlew spotlessApply :plugins:tms:test -PskipITs
```
---------
Co-authored-by: Cursor <[email protected]>
---
build.gradle.kts | 3 +-
plugins/tms/build.gradle.kts | 76 +++++++++
.../mapper/TableMaintenanceActionMapper.java | 67 ++++++++
.../TableMaintenanceActionSQLProviderFactory.java | 131 +++++++++++++++
.../storage/mapper/TableSnapshotMetricsMapper.java | 41 +++++
.../TableSnapshotMetricsSQLProviderFactory.java | 80 +++++++++
.../mapper/provider/TmsMapperPackageProvider.java | 34 ++++
.../TableMaintenanceActionBaseSQLProvider.java | 128 +++++++++++++++
.../base/TableSnapshotMetricsBaseSQLProvider.java | 56 +++++++
.../h2/TableMaintenanceActionH2Provider.java | 24 +++
.../h2/TableSnapshotMetricsH2Provider.java | 24 +++
.../TableMaintenanceActionPostgreSQLProvider.java | 25 +++
.../TableSnapshotMetricsPostgreSQLProvider.java | 24 +++
.../storage/po/TableMaintenanceActionPO.java | 48 ++++++
.../storage/po/TableSnapshotMetricsPO.java | 46 ++++++
.../service/TableMaintenanceActionMetaService.java | 178 +++++++++++++++++++++
.../service/TableSnapshotMetricsMetaService.java | 87 ++++++++++
...elational.mapper.provider.MapperPackageProvider | 18 +++
.../mapper/TestTmsMapperPackageProvider.java | 51 ++++++
.../TestTableMaintenanceActionMetaService.java | 65 ++++++++
.../TestTableSnapshotMetricsMetaService.java | 46 ++++++
scripts/h2/schema-2.0.0-h2.sql | 29 ++++
scripts/h2/upgrade-1.3.0-to-2.0.0-h2.sql | 29 ++++
scripts/mysql/schema-2.0.0-mysql.sql | 31 ++++
scripts/mysql/upgrade-1.3.0-to-2.0.0-mysql.sql | 31 ++++
scripts/postgresql/schema-2.0.0-postgresql.sql | 43 +++++
.../upgrade-1.3.0-to-2.0.0-postgresql.sql | 43 +++++
settings.gradle.kts | 1 +
28 files changed, 1458 insertions(+), 1 deletion(-)
diff --git a/build.gradle.kts b/build.gradle.kts
index f6bf7855ea..0161f01f72 100644
--- a/build.gradle.kts
+++ b/build.gradle.kts
@@ -1170,7 +1170,8 @@ tasks {
":iceberg:iceberg-rest-server:copyLibAndConfigs",
":lance:lance-rest-server:copyLibAndConfigs",
":maintenance:optimizer:copyLibAndConfigs",
- ":plugins:idp-basic:copyLibAndConfigs"
+ ":plugins:idp-basic:copyLibAndConfigs",
+ ":plugins:tms:copyLibAndConfigs"
)
if (!skipWeb) {
dependencies.add(":web:web:build")
diff --git a/plugins/tms/build.gradle.kts b/plugins/tms/build.gradle.kts
new file mode 100644
index 0000000000..1e6d92d9ad
--- /dev/null
+++ b/plugins/tms/build.gradle.kts
@@ -0,0 +1,76 @@
+/*
+ * 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.
+ */
+
+plugins {
+ `maven-publish`
+ id("java")
+ id("idea")
+}
+
+dependencies {
+ annotationProcessor(libs.lombok)
+
+ implementation(project(":api"))
+ implementation(project(":common"))
+ implementation(project(":core"))
+
+ implementation(libs.guava)
+ implementation(libs.mybatis)
+
+ compileOnly(libs.lombok)
+ compileOnly(libs.slf4j.api)
+
+ testImplementation(project(":clients:client-java"))
+ testImplementation(project(":core", "testArtifacts"))
+ testImplementation(project(":integration-test-common", "testArtifacts"))
+ testImplementation(project(":server"))
+ testImplementation(project(":server-common"))
+ testImplementation(libs.junit.jupiter.api)
+ testImplementation(libs.junit.jupiter.params)
+ testImplementation(libs.mockito.core)
+ testImplementation(libs.mysql.driver)
+ testImplementation(libs.postgresql.driver)
+ testImplementation(libs.testcontainers)
+ testImplementation(libs.testcontainers.mysql)
+ testImplementation(libs.testcontainers.postgresql)
+ testRuntimeOnly(libs.junit.jupiter.engine)
+}
+
+tasks {
+ val copyLibs by registering(Copy::class) {
+ dependsOn(jar)
+ from(layout.buildDirectory.dir("libs")) {
+ include("gravitino-tms-*.jar")
+ exclude("*-javadoc.jar", "*-sources.jar")
+ }
+ into("$rootDir/distribution/package/libs")
+ duplicatesStrategy = DuplicatesStrategy.EXCLUDE
+ }
+
+ register("copyLibAndConfigs", Copy::class) {
+ group = "gravitino distribution"
+ description = "Copy tms plugin jar into distribution package libs"
+ dependsOn(copyLibs)
+ }
+
+ test {
+ useJUnitPlatform()
+ environment("GRAVITINO_HOME", rootDir.path)
+ }
+}
diff --git
a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/TableMaintenanceActionMapper.java
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/TableMaintenanceActionMapper.java
new file mode 100644
index 0000000000..ac14efa56a
--- /dev/null
+++
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/TableMaintenanceActionMapper.java
@@ -0,0 +1,67 @@
+/*
+ * 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.maintenance.storage.mapper;
+
+import javax.annotation.Nullable;
+import org.apache.gravitino.maintenance.storage.po.TableMaintenanceActionPO;
+import org.apache.ibatis.annotations.DeleteProvider;
+import org.apache.ibatis.annotations.InsertProvider;
+import org.apache.ibatis.annotations.Options;
+import org.apache.ibatis.annotations.Param;
+import org.apache.ibatis.annotations.SelectProvider;
+import org.apache.ibatis.annotations.UpdateProvider;
+
+/** MyBatis mapper for {@code table_maintenance_action}. */
+public interface TableMaintenanceActionMapper {
+
+ String TABLE_NAME = "table_maintenance_action";
+
+ @InsertProvider(type = TableMaintenanceActionSQLProviderFactory.class,
method = "insertAction")
+ @Options(useGeneratedKeys = true, keyProperty = "actionMeta.id", keyColumn =
"id")
+ void insertAction(@Param("actionMeta") TableMaintenanceActionPO actionMeta);
+
+ @UpdateProvider(type = TableMaintenanceActionSQLProviderFactory.class,
method = "bindJobId")
+ int bindJobId(@Param("id") long id, @Param("jobId") long jobId);
+
+ @Nullable
+ @SelectProvider(type = TableMaintenanceActionSQLProviderFactory.class,
method = "selectById")
+ TableMaintenanceActionPO selectById(@Param("id") long id);
+
+ @Nullable
+ @SelectProvider(type = TableMaintenanceActionSQLProviderFactory.class,
method = "selectByJobId")
+ TableMaintenanceActionPO selectByJobId(@Param("jobId") long jobId);
+
+ @UpdateProvider(
+ type = TableMaintenanceActionSQLProviderFactory.class,
+ method = "updateBeforeSnapshot")
+ int updateBeforeSnapshot(
+ @Param("jobId") long jobId, @Param("beforeSnapshotId") long
beforeSnapshotId);
+
+ @UpdateProvider(type = TableMaintenanceActionSQLProviderFactory.class,
method = "markFinished")
+ int markFinished(
+ @Param("jobId") long jobId,
+ @Param("afterSnapshotId") Long afterSnapshotId,
+ @Param("finishedAt") long finishedAt);
+
+ @DeleteProvider(type = TableMaintenanceActionSQLProviderFactory.class,
method = "deleteById")
+ int deleteById(@Param("id") long id);
+
+ @DeleteProvider(type = TableMaintenanceActionSQLProviderFactory.class,
method = "deleteByJobId")
+ int deleteByJobId(@Param("jobId") long jobId);
+}
diff --git
a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/TableMaintenanceActionSQLProviderFactory.java
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/TableMaintenanceActionSQLProviderFactory.java
new file mode 100644
index 0000000000..2bfe0b66ee
--- /dev/null
+++
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/TableMaintenanceActionSQLProviderFactory.java
@@ -0,0 +1,131 @@
+/*
+ * 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.maintenance.storage.mapper;
+
+import com.google.common.collect.ImmutableMap;
+import java.util.Map;
+import
org.apache.gravitino.maintenance.storage.mapper.provider.base.TableMaintenanceActionBaseSQLProvider;
+import
org.apache.gravitino.maintenance.storage.mapper.provider.h2.TableMaintenanceActionH2Provider;
+import
org.apache.gravitino.maintenance.storage.mapper.provider.postgresql.TableMaintenanceActionPostgreSQLProvider;
+import org.apache.gravitino.maintenance.storage.po.TableMaintenanceActionPO;
+import org.apache.gravitino.storage.relational.JDBCBackend.JDBCBackendType;
+import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
+import org.apache.ibatis.annotations.Param;
+
+/** SQL provider factory for {@link TableMaintenanceActionMapper}. */
+public class TableMaintenanceActionSQLProviderFactory {
+
+ private static final TableMaintenanceActionBaseSQLProvider MYSQL_PROVIDER =
+ new TableMaintenanceActionBaseSQLProvider();
+ private static final TableMaintenanceActionBaseSQLProvider H2_PROVIDER =
+ new TableMaintenanceActionH2Provider();
+ private static final TableMaintenanceActionBaseSQLProvider
POSTGRESQL_PROVIDER =
+ new TableMaintenanceActionPostgreSQLProvider();
+ private static final Map<JDBCBackendType,
TableMaintenanceActionBaseSQLProvider> PROVIDER_MAP =
+ ImmutableMap.of(
+ JDBCBackendType.MYSQL,
+ MYSQL_PROVIDER,
+ JDBCBackendType.H2,
+ H2_PROVIDER,
+ JDBCBackendType.POSTGRESQL,
+ POSTGRESQL_PROVIDER);
+
+ private TableMaintenanceActionSQLProviderFactory() {}
+
+ /**
+ * @param actionMeta row to insert
+ * @return INSERT statement
+ */
+ public static String insertAction(@Param("actionMeta")
TableMaintenanceActionPO actionMeta) {
+ return getProvider().insertAction(actionMeta);
+ }
+
+ /**
+ * @param id action id
+ * @param jobId job run id
+ * @return UPDATE statement
+ */
+ public static String bindJobId(@Param("id") long id, @Param("jobId") long
jobId) {
+ return getProvider().bindJobId(id, jobId);
+ }
+
+ /**
+ * @param id action id
+ * @return SELECT statement
+ */
+ public static String selectById(@Param("id") long id) {
+ return getProvider().selectById(id);
+ }
+
+ /**
+ * @param jobId job run id
+ * @return SELECT statement
+ */
+ public static String selectByJobId(@Param("jobId") long jobId) {
+ return getProvider().selectByJobId(jobId);
+ }
+
+ /**
+ * @param jobId job run id
+ * @param beforeSnapshotId snapshot id
+ * @return UPDATE statement
+ */
+ public static String updateBeforeSnapshot(
+ @Param("jobId") long jobId, @Param("beforeSnapshotId") long
beforeSnapshotId) {
+ return getProvider().updateBeforeSnapshot(jobId, beforeSnapshotId);
+ }
+
+ /**
+ * @param jobId job run id
+ * @param afterSnapshotId after snapshot
+ * @param finishedAt finish millis
+ * @return UPDATE statement
+ */
+ public static String markFinished(
+ @Param("jobId") long jobId,
+ @Param("afterSnapshotId") Long afterSnapshotId,
+ @Param("finishedAt") long finishedAt) {
+ return getProvider().markFinished(jobId, afterSnapshotId, finishedAt);
+ }
+
+ /**
+ * @param id action id
+ * @return DELETE statement
+ */
+ public static String deleteById(@Param("id") long id) {
+ return getProvider().deleteById(id);
+ }
+
+ /**
+ * @param jobId job run id
+ * @return DELETE statement
+ */
+ public static String deleteByJobId(@Param("jobId") long jobId) {
+ return getProvider().deleteByJobId(jobId);
+ }
+
+ static TableMaintenanceActionBaseSQLProvider getProvider() {
+ String databaseId =
+ SqlSessionFactoryHelper.getInstance()
+ .getSqlSessionFactory()
+ .getConfiguration()
+ .getDatabaseId();
+ return PROVIDER_MAP.get(JDBCBackendType.fromString(databaseId));
+ }
+}
diff --git
a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/TableSnapshotMetricsMapper.java
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/TableSnapshotMetricsMapper.java
new file mode 100644
index 0000000000..d80303fc1f
--- /dev/null
+++
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/TableSnapshotMetricsMapper.java
@@ -0,0 +1,41 @@
+/*
+ * 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.maintenance.storage.mapper;
+
+import javax.annotation.Nullable;
+import org.apache.gravitino.maintenance.storage.po.TableSnapshotMetricsPO;
+import org.apache.ibatis.annotations.InsertProvider;
+import org.apache.ibatis.annotations.Param;
+import org.apache.ibatis.annotations.SelectProvider;
+
+/** MyBatis mapper for {@code table_snapshot_metrics}. */
+public interface TableSnapshotMetricsMapper {
+
+ String TABLE_NAME = "table_snapshot_metrics";
+
+ @InsertProvider(type = TableSnapshotMetricsSQLProviderFactory.class, method
= "insertMetrics")
+ void insertMetrics(@Param("metricsMeta") TableSnapshotMetricsPO metricsMeta);
+
+ @Nullable
+ @SelectProvider(type = TableSnapshotMetricsSQLProviderFactory.class, method
= "selectMetrics")
+ TableSnapshotMetricsPO selectMetrics(
+ @Param("tableId") long tableId,
+ @Param("policyId") long policyId,
+ @Param("snapshotId") long snapshotId);
+}
diff --git
a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/TableSnapshotMetricsSQLProviderFactory.java
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/TableSnapshotMetricsSQLProviderFactory.java
new file mode 100644
index 0000000000..1ab21cd27f
--- /dev/null
+++
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/TableSnapshotMetricsSQLProviderFactory.java
@@ -0,0 +1,80 @@
+/*
+ * 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.maintenance.storage.mapper;
+
+import com.google.common.collect.ImmutableMap;
+import java.util.Map;
+import
org.apache.gravitino.maintenance.storage.mapper.provider.base.TableSnapshotMetricsBaseSQLProvider;
+import
org.apache.gravitino.maintenance.storage.mapper.provider.h2.TableSnapshotMetricsH2Provider;
+import
org.apache.gravitino.maintenance.storage.mapper.provider.postgresql.TableSnapshotMetricsPostgreSQLProvider;
+import org.apache.gravitino.maintenance.storage.po.TableSnapshotMetricsPO;
+import org.apache.gravitino.storage.relational.JDBCBackend.JDBCBackendType;
+import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
+import org.apache.ibatis.annotations.Param;
+
+/** SQL provider factory for {@link TableSnapshotMetricsMapper}. */
+public class TableSnapshotMetricsSQLProviderFactory {
+
+ private static final TableSnapshotMetricsBaseSQLProvider MYSQL_PROVIDER =
+ new TableSnapshotMetricsBaseSQLProvider();
+ private static final TableSnapshotMetricsBaseSQLProvider H2_PROVIDER =
+ new TableSnapshotMetricsH2Provider();
+ private static final TableSnapshotMetricsBaseSQLProvider POSTGRESQL_PROVIDER
=
+ new TableSnapshotMetricsPostgreSQLProvider();
+ private static final Map<JDBCBackendType,
TableSnapshotMetricsBaseSQLProvider> PROVIDER_MAP =
+ ImmutableMap.of(
+ JDBCBackendType.MYSQL,
+ MYSQL_PROVIDER,
+ JDBCBackendType.H2,
+ H2_PROVIDER,
+ JDBCBackendType.POSTGRESQL,
+ POSTGRESQL_PROVIDER);
+
+ private TableSnapshotMetricsSQLProviderFactory() {}
+
+ /**
+ * @param metricsMeta metrics row
+ * @return INSERT statement
+ */
+ public static String insertMetrics(@Param("metricsMeta")
TableSnapshotMetricsPO metricsMeta) {
+ return getProvider().insertMetrics(metricsMeta);
+ }
+
+ /**
+ * @param tableId table id
+ * @param policyId policy id
+ * @param snapshotId snapshot id
+ * @return SELECT statement
+ */
+ public static String selectMetrics(
+ @Param("tableId") long tableId,
+ @Param("policyId") long policyId,
+ @Param("snapshotId") long snapshotId) {
+ return getProvider().selectMetrics(tableId, policyId, snapshotId);
+ }
+
+ static TableSnapshotMetricsBaseSQLProvider getProvider() {
+ String databaseId =
+ SqlSessionFactoryHelper.getInstance()
+ .getSqlSessionFactory()
+ .getConfiguration()
+ .getDatabaseId();
+ return PROVIDER_MAP.get(JDBCBackendType.fromString(databaseId));
+ }
+}
diff --git
a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/provider/TmsMapperPackageProvider.java
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/provider/TmsMapperPackageProvider.java
new file mode 100644
index 0000000000..bafb8ce6e3
--- /dev/null
+++
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/provider/TmsMapperPackageProvider.java
@@ -0,0 +1,34 @@
+/*
+ * 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.maintenance.storage.mapper.provider;
+
+import com.google.common.collect.ImmutableList;
+import java.util.List;
+import
org.apache.gravitino.maintenance.storage.mapper.TableMaintenanceActionMapper;
+import
org.apache.gravitino.maintenance.storage.mapper.TableSnapshotMetricsMapper;
+import
org.apache.gravitino.storage.relational.mapper.provider.MapperPackageProvider;
+
+/** Supplies TMS mapper classes from the {@code tms} plugin. */
+public class TmsMapperPackageProvider implements MapperPackageProvider {
+
+ @Override
+ public List<Class<?>> getMapperClasses() {
+ return ImmutableList.of(TableMaintenanceActionMapper.class,
TableSnapshotMetricsMapper.class);
+ }
+}
diff --git
a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/provider/base/TableMaintenanceActionBaseSQLProvider.java
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/provider/base/TableMaintenanceActionBaseSQLProvider.java
new file mode 100644
index 0000000000..9b5c44b134
--- /dev/null
+++
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/provider/base/TableMaintenanceActionBaseSQLProvider.java
@@ -0,0 +1,128 @@
+/*
+ * 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.maintenance.storage.mapper.provider.base;
+
+import static
org.apache.gravitino.maintenance.storage.mapper.TableMaintenanceActionMapper.TABLE_NAME;
+
+import org.apache.gravitino.maintenance.storage.po.TableMaintenanceActionPO;
+import org.apache.ibatis.annotations.Param;
+
+/** Portable SQL for {@code table_maintenance_action}. */
+public class TableMaintenanceActionBaseSQLProvider {
+
+ /**
+ * Inserts an in-flight occupancy row before {@code runJob}. {@code job_id}
stays null until bound
+ * after submission (§5.2).
+ *
+ * @param actionMeta row to insert
+ * @return INSERT statement
+ */
+ public String insertAction(@Param("actionMeta") TableMaintenanceActionPO
actionMeta) {
+ return "INSERT INTO "
+ + TABLE_NAME
+ + " (table_id, policy_id, created_at)"
+ + " VALUES (#{actionMeta.tableId}, #{actionMeta.policyId},
#{actionMeta.createdAt})";
+ }
+
+ /**
+ * Binds the Gravitino job run id after {@code runJob} succeeds.
+ *
+ * @param id action row id
+ * @param jobId {@code job_run_meta.job_run_id}
+ * @return UPDATE statement
+ */
+ public String bindJobId(@Param("id") long id, @Param("jobId") long jobId) {
+ return "UPDATE "
+ + TABLE_NAME
+ + " SET job_id = #{jobId}"
+ + " WHERE id = #{id} AND job_id IS NULL AND finished_at IS NULL";
+ }
+
+ /**
+ * @param id action row id
+ * @return SELECT statement
+ */
+ public String selectById(@Param("id") long id) {
+ return selectColumns() + " WHERE id = #{id}";
+ }
+
+ /**
+ * @param jobId Gravitino job run id
+ * @return SELECT statement
+ */
+ public String selectByJobId(@Param("jobId") long jobId) {
+ return selectColumns() + " WHERE job_id = #{jobId}";
+ }
+
+ /**
+ * Sets {@code before_snapshot_id} once when {@code S0} is sampled.
+ *
+ * @param jobId Gravitino job run id
+ * @param beforeSnapshotId sampled snapshot id
+ * @return UPDATE statement
+ */
+ public String updateBeforeSnapshot(
+ @Param("jobId") long jobId, @Param("beforeSnapshotId") long
beforeSnapshotId) {
+ return "UPDATE "
+ + TABLE_NAME
+ + " SET before_snapshot_id = #{beforeSnapshotId}"
+ + " WHERE job_id = #{jobId} AND before_snapshot_id IS NULL AND
finished_at IS NULL";
+ }
+
+ /**
+ * Sets terminal snapshot pointers when the Spark Job finishes.
+ *
+ * @param jobId Gravitino job run id
+ * @param afterSnapshotId terminal snapshot id
+ * @param finishedAt finish time millis
+ * @return UPDATE statement
+ */
+ public String markFinished(
+ @Param("jobId") long jobId,
+ @Param("afterSnapshotId") Long afterSnapshotId,
+ @Param("finishedAt") long finishedAt) {
+ return "UPDATE "
+ + TABLE_NAME
+ + " SET after_snapshot_id = #{afterSnapshotId}, finished_at =
#{finishedAt}"
+ + " WHERE job_id = #{jobId} AND finished_at IS NULL";
+ }
+
+ /**
+ * @param id action row id
+ * @return DELETE statement
+ */
+ public String deleteById(@Param("id") long id) {
+ return "DELETE FROM " + TABLE_NAME + " WHERE id = #{id}";
+ }
+
+ /**
+ * @param jobId Gravitino job run id
+ * @return DELETE statement
+ */
+ public String deleteByJobId(@Param("jobId") long jobId) {
+ return "DELETE FROM " + TABLE_NAME + " WHERE job_id = #{jobId}";
+ }
+
+ private static String selectColumns() {
+ return "SELECT id, job_id AS jobId, table_id AS tableId, policy_id AS
policyId,"
+ + " before_snapshot_id AS beforeSnapshotId, after_snapshot_id AS
afterSnapshotId,"
+ + " finished_at AS finishedAt, created_at AS createdAt FROM "
+ + TABLE_NAME;
+ }
+}
diff --git
a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/provider/base/TableSnapshotMetricsBaseSQLProvider.java
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/provider/base/TableSnapshotMetricsBaseSQLProvider.java
new file mode 100644
index 0000000000..b3b58ac2fa
--- /dev/null
+++
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/provider/base/TableSnapshotMetricsBaseSQLProvider.java
@@ -0,0 +1,56 @@
+/*
+ * 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.maintenance.storage.mapper.provider.base;
+
+import
org.apache.gravitino.maintenance.storage.mapper.TableSnapshotMetricsMapper;
+import org.apache.gravitino.maintenance.storage.po.TableSnapshotMetricsPO;
+import org.apache.ibatis.annotations.Param;
+
+/** Portable SQL for {@code table_snapshot_metrics}. */
+public class TableSnapshotMetricsBaseSQLProvider {
+
+ /**
+ * @param metricsMeta metrics row
+ * @return INSERT statement
+ */
+ public String insertMetrics(@Param("metricsMeta") TableSnapshotMetricsPO
metricsMeta) {
+ return "INSERT INTO "
+ + TableSnapshotMetricsMapper.TABLE_NAME
+ + " (table_id, policy_id, snapshot_id, metrics_value, created_at)"
+ + " VALUES (#{metricsMeta.tableId}, #{metricsMeta.policyId},
#{metricsMeta.snapshotId},"
+ + " #{metricsMeta.metricsValue}, #{metricsMeta.createdAt})";
+ }
+
+ /**
+ * @param tableId table id
+ * @param policyId policy id
+ * @param snapshotId Iceberg snapshot id
+ * @return SELECT statement
+ */
+ public String selectMetrics(
+ @Param("tableId") long tableId,
+ @Param("policyId") long policyId,
+ @Param("snapshotId") long snapshotId) {
+ return "SELECT id, table_id AS tableId, policy_id AS policyId, snapshot_id
AS snapshotId,"
+ + " metrics_value AS metricsValue, created_at AS createdAt FROM "
+ + TableSnapshotMetricsMapper.TABLE_NAME
+ + " WHERE table_id = #{tableId} AND policy_id = #{policyId} AND
snapshot_id ="
+ + " #{snapshotId}";
+ }
+}
diff --git
a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/provider/h2/TableMaintenanceActionH2Provider.java
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/provider/h2/TableMaintenanceActionH2Provider.java
new file mode 100644
index 0000000000..835f656b70
--- /dev/null
+++
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/provider/h2/TableMaintenanceActionH2Provider.java
@@ -0,0 +1,24 @@
+/*
+ * 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.maintenance.storage.mapper.provider.h2;
+
+import
org.apache.gravitino.maintenance.storage.mapper.provider.base.TableMaintenanceActionBaseSQLProvider;
+
+/** H2 SQL provider for {@code table_maintenance_action}. */
+public class TableMaintenanceActionH2Provider extends
TableMaintenanceActionBaseSQLProvider {}
diff --git
a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/provider/h2/TableSnapshotMetricsH2Provider.java
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/provider/h2/TableSnapshotMetricsH2Provider.java
new file mode 100644
index 0000000000..7a532055c5
--- /dev/null
+++
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/provider/h2/TableSnapshotMetricsH2Provider.java
@@ -0,0 +1,24 @@
+/*
+ * 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.maintenance.storage.mapper.provider.h2;
+
+import
org.apache.gravitino.maintenance.storage.mapper.provider.base.TableSnapshotMetricsBaseSQLProvider;
+
+/** H2 SQL provider for {@code table_snapshot_metrics}. */
+public class TableSnapshotMetricsH2Provider extends
TableSnapshotMetricsBaseSQLProvider {}
diff --git
a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/provider/postgresql/TableMaintenanceActionPostgreSQLProvider.java
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/provider/postgresql/TableMaintenanceActionPostgreSQLProvider.java
new file mode 100644
index 0000000000..49f89e7827
--- /dev/null
+++
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/provider/postgresql/TableMaintenanceActionPostgreSQLProvider.java
@@ -0,0 +1,25 @@
+/*
+ * 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.maintenance.storage.mapper.provider.postgresql;
+
+import
org.apache.gravitino.maintenance.storage.mapper.provider.base.TableMaintenanceActionBaseSQLProvider;
+
+/** PostgreSQL SQL provider for {@code table_maintenance_action}. */
+public class TableMaintenanceActionPostgreSQLProvider
+ extends TableMaintenanceActionBaseSQLProvider {}
diff --git
a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/provider/postgresql/TableSnapshotMetricsPostgreSQLProvider.java
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/provider/postgresql/TableSnapshotMetricsPostgreSQLProvider.java
new file mode 100644
index 0000000000..16484522f7
--- /dev/null
+++
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/provider/postgresql/TableSnapshotMetricsPostgreSQLProvider.java
@@ -0,0 +1,24 @@
+/*
+ * 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.maintenance.storage.mapper.provider.postgresql;
+
+import
org.apache.gravitino.maintenance.storage.mapper.provider.base.TableSnapshotMetricsBaseSQLProvider;
+
+/** PostgreSQL SQL provider for {@code table_snapshot_metrics}. */
+public class TableSnapshotMetricsPostgreSQLProvider extends
TableSnapshotMetricsBaseSQLProvider {}
diff --git
a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/po/TableMaintenanceActionPO.java
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/po/TableMaintenanceActionPO.java
new file mode 100644
index 0000000000..1f1928bbae
--- /dev/null
+++
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/po/TableMaintenanceActionPO.java
@@ -0,0 +1,48 @@
+/*
+ * 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.maintenance.storage.po;
+
+import lombok.AccessLevel;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.ToString;
+import lombok.experimental.Accessors;
+
+/** Persistence object for {@code table_maintenance_action}. */
+@Getter
+@Accessors(fluent = true)
+@EqualsAndHashCode
+@ToString
+@Builder(setterPrefix = "with")
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+@AllArgsConstructor(access = AccessLevel.PRIVATE)
+public class TableMaintenanceActionPO {
+
+ private Long id;
+ private Long jobId;
+ private Long tableId;
+ private Long policyId;
+ private Long beforeSnapshotId;
+ private Long afterSnapshotId;
+ private Long finishedAt;
+ private Long createdAt;
+}
diff --git
a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/po/TableSnapshotMetricsPO.java
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/po/TableSnapshotMetricsPO.java
new file mode 100644
index 0000000000..8cb2fd1313
--- /dev/null
+++
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/po/TableSnapshotMetricsPO.java
@@ -0,0 +1,46 @@
+/*
+ * 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.maintenance.storage.po;
+
+import lombok.AccessLevel;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.ToString;
+import lombok.experimental.Accessors;
+
+/** Persistence object for {@code table_snapshot_metrics}. */
+@Getter
+@Accessors(fluent = true)
+@EqualsAndHashCode
+@ToString
+@Builder(setterPrefix = "with")
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+@AllArgsConstructor(access = AccessLevel.PRIVATE)
+public class TableSnapshotMetricsPO {
+
+ private Long id;
+ private Long tableId;
+ private Long policyId;
+ private Long snapshotId;
+ private String metricsValue;
+ private Long createdAt;
+}
diff --git
a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/service/TableMaintenanceActionMetaService.java
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/service/TableMaintenanceActionMetaService.java
new file mode 100644
index 0000000000..ea5c570d3e
--- /dev/null
+++
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/service/TableMaintenanceActionMetaService.java
@@ -0,0 +1,178 @@
+/*
+ * 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.maintenance.storage.service;
+
+import static
org.apache.gravitino.metrics.source.MetricsSource.GRAVITINO_RELATIONAL_STORE_METRIC_NAME;
+
+import com.google.common.base.Preconditions;
+import javax.annotation.Nullable;
+import
org.apache.gravitino.maintenance.storage.mapper.TableMaintenanceActionMapper;
+import org.apache.gravitino.maintenance.storage.po.TableMaintenanceActionPO;
+import org.apache.gravitino.metrics.Monitored;
+import org.apache.gravitino.storage.relational.utils.SessionUtils;
+
+/** DAO service for {@code table_maintenance_action}. */
+public class TableMaintenanceActionMetaService {
+
+ private static final TableMaintenanceActionMetaService INSTANCE =
+ new TableMaintenanceActionMetaService();
+
+ private TableMaintenanceActionMetaService() {}
+
+ /**
+ * @return singleton service instance
+ */
+ public static TableMaintenanceActionMetaService getInstance() {
+ return INSTANCE;
+ }
+
+ /**
+ * Inserts an in-flight occupancy row before {@code runJob}. {@code job_id}
is bound later via
+ * {@link #bindJobId(long, long)}.
+ *
+ * @param tableId table entity id
+ * @param policyId policy entity id
+ * @return generated action id
+ */
+ @Monitored(
+ metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
+ baseMetricName = "insertTableMaintenanceAction")
+ public long insertInFlight(long tableId, long policyId) {
+ TableMaintenanceActionPO actionMeta =
+ TableMaintenanceActionPO.builder()
+ .withTableId(tableId)
+ .withPolicyId(policyId)
+ .withCreatedAt(System.currentTimeMillis())
+ .build();
+ SessionUtils.doWithCommit(
+ TableMaintenanceActionMapper.class, mapper ->
mapper.insertAction(actionMeta));
+ Preconditions.checkState(actionMeta.id() != null, "insert did not return
generated id");
+ return actionMeta.id();
+ }
+
+ /**
+ * Binds {@code job_run_meta.job_run_id} after a successful {@code runJob}.
+ *
+ * @param id action row id
+ * @param jobId Gravitino job run id
+ */
+ @Monitored(
+ metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
+ baseMetricName = "bindTableMaintenanceActionJobId")
+ public void bindJobId(long id, long jobId) {
+ int updated =
+ SessionUtils.doWithCommitAndFetchResult(
+ TableMaintenanceActionMapper.class, mapper -> mapper.bindJobId(id,
jobId));
+ Preconditions.checkState(
+ updated == 1,
+ "expected to bind job_id on exactly one in-flight action, updated=%s",
+ updated);
+ }
+
+ /**
+ * @param id action row id
+ * @return action row or null
+ */
+ @Nullable
+ @Monitored(
+ metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
+ baseMetricName = "getTableMaintenanceActionById")
+ public TableMaintenanceActionPO getById(long id) {
+ return SessionUtils.getWithoutCommit(
+ TableMaintenanceActionMapper.class, mapper -> mapper.selectById(id));
+ }
+
+ /**
+ * @param jobId Gravitino job run id
+ * @return action row or null
+ */
+ @Nullable
+ @Monitored(
+ metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
+ baseMetricName = "getTableMaintenanceActionByJobId")
+ public TableMaintenanceActionPO getByJobId(long jobId) {
+ return SessionUtils.getWithoutCommit(
+ TableMaintenanceActionMapper.class, mapper ->
mapper.selectByJobId(jobId));
+ }
+
+ /**
+ * Records the sampled Iceberg snapshot id ({@code S0}) once for a running
action.
+ *
+ * @param jobId Gravitino job run id
+ * @param beforeSnapshotId sampled snapshot id
+ */
+ @Monitored(
+ metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
+ baseMetricName = "recordTableMaintenanceActionBeforeSnapshot")
+ public void recordBeforeSnapshot(long jobId, long beforeSnapshotId) {
+ int updated =
+ SessionUtils.doWithCommitAndFetchResult(
+ TableMaintenanceActionMapper.class,
+ mapper -> mapper.updateBeforeSnapshot(jobId, beforeSnapshotId));
+ Preconditions.checkState(
+ updated == 1,
+ "expected to set before_snapshot_id once on an unfinished action,
updated=%s",
+ updated);
+ }
+
+ /**
+ * Marks the action finished with the terminal snapshot id.
+ *
+ * @param jobId Gravitino job run id
+ * @param afterSnapshotId terminal snapshot id
+ * @param finishedAt finish timestamp millis
+ */
+ @Monitored(
+ metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
+ baseMetricName = "markTableMaintenanceActionFinished")
+ public void markFinished(long jobId, Long afterSnapshotId, long finishedAt) {
+ int updated =
+ SessionUtils.doWithCommitAndFetchResult(
+ TableMaintenanceActionMapper.class,
+ mapper -> mapper.markFinished(jobId, afterSnapshotId, finishedAt));
+ Preconditions.checkState(
+ updated == 1, "expected to finish exactly one in-flight action,
updated=%s", updated);
+ }
+
+ /**
+ * Deletes the {@code table_maintenance_action} row. Does not delete {@code
+ * table_snapshot_metrics}; metrics are retained for TTL/N cleanup.
+ *
+ * @param id action row id
+ */
+ @Monitored(
+ metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
+ baseMetricName = "deleteTableMaintenanceActionById")
+ public void deleteById(long id) {
+ SessionUtils.doWithCommit(TableMaintenanceActionMapper.class, mapper ->
mapper.deleteById(id));
+ }
+
+ /**
+ * Deletes the {@code table_maintenance_action} row keyed by job run id.
Does not delete metrics.
+ *
+ * @param jobId Gravitino job run id
+ */
+ @Monitored(
+ metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
+ baseMetricName = "deleteTableMaintenanceActionByJobId")
+ public void deleteByJobId(long jobId) {
+ SessionUtils.doWithCommit(
+ TableMaintenanceActionMapper.class, mapper ->
mapper.deleteByJobId(jobId));
+ }
+}
diff --git
a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/service/TableSnapshotMetricsMetaService.java
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/service/TableSnapshotMetricsMetaService.java
new file mode 100644
index 0000000000..392c846b94
--- /dev/null
+++
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/service/TableSnapshotMetricsMetaService.java
@@ -0,0 +1,87 @@
+/*
+ * 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.maintenance.storage.service;
+
+import static
org.apache.gravitino.metrics.source.MetricsSource.GRAVITINO_RELATIONAL_STORE_METRIC_NAME;
+
+import javax.annotation.Nullable;
+import
org.apache.gravitino.maintenance.storage.mapper.TableSnapshotMetricsMapper;
+import org.apache.gravitino.maintenance.storage.po.TableSnapshotMetricsPO;
+import org.apache.gravitino.metrics.Monitored;
+import org.apache.gravitino.storage.relational.utils.SessionUtils;
+
+/** DAO service for {@code table_snapshot_metrics}. */
+public class TableSnapshotMetricsMetaService {
+
+ private static final TableSnapshotMetricsMetaService INSTANCE =
+ new TableSnapshotMetricsMetaService();
+
+ private TableSnapshotMetricsMetaService() {}
+
+ /**
+ * @return singleton service instance
+ */
+ public static TableSnapshotMetricsMetaService getInstance() {
+ return INSTANCE;
+ }
+
+ /**
+ * Idempotently stores metrics JSON for a table/policy/snapshot triple. If a
row already exists,
+ * this is a no-op (does not overwrite) so sample / listener retries are
safe.
+ *
+ * @param tableId table id
+ * @param policyId policy id
+ * @param snapshotId Iceberg snapshot id
+ * @param metricsJson serialized metrics payload
+ */
+ @Monitored(
+ metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
+ baseMetricName = "insertTableSnapshotMetrics")
+ public void insertMetrics(long tableId, long policyId, long snapshotId,
String metricsJson) {
+ if (getMetrics(tableId, policyId, snapshotId) != null) {
+ return;
+ }
+ TableSnapshotMetricsPO metricsMeta =
+ TableSnapshotMetricsPO.builder()
+ .withTableId(tableId)
+ .withPolicyId(policyId)
+ .withSnapshotId(snapshotId)
+ .withMetricsValue(metricsJson)
+ .withCreatedAt(System.currentTimeMillis())
+ .build();
+ SessionUtils.doWithCommit(
+ TableSnapshotMetricsMapper.class, mapper ->
mapper.insertMetrics(metricsMeta));
+ }
+
+ /**
+ * @param tableId table id
+ * @param policyId policy id
+ * @param snapshotId snapshot id
+ * @return stored metrics or null
+ */
+ @Nullable
+ @Monitored(
+ metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
+ baseMetricName = "getTableSnapshotMetrics")
+ public TableSnapshotMetricsPO getMetrics(long tableId, long policyId, long
snapshotId) {
+ return SessionUtils.getWithoutCommit(
+ TableSnapshotMetricsMapper.class,
+ mapper -> mapper.selectMetrics(tableId, policyId, snapshotId));
+ }
+}
diff --git
a/plugins/tms/src/main/resources/META-INF/services/org.apache.gravitino.storage.relational.mapper.provider.MapperPackageProvider
b/plugins/tms/src/main/resources/META-INF/services/org.apache.gravitino.storage.relational.mapper.provider.MapperPackageProvider
new file mode 100644
index 0000000000..a130adea4e
--- /dev/null
+++
b/plugins/tms/src/main/resources/META-INF/services/org.apache.gravitino.storage.relational.mapper.provider.MapperPackageProvider
@@ -0,0 +1,18 @@
+# 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.maintenance.storage.mapper.provider.TmsMapperPackageProvider
diff --git
a/plugins/tms/src/test/java/org/apache/gravitino/maintenance/storage/mapper/TestTmsMapperPackageProvider.java
b/plugins/tms/src/test/java/org/apache/gravitino/maintenance/storage/mapper/TestTmsMapperPackageProvider.java
new file mode 100644
index 0000000000..f0afbe27fb
--- /dev/null
+++
b/plugins/tms/src/test/java/org/apache/gravitino/maintenance/storage/mapper/TestTmsMapperPackageProvider.java
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.gravitino.maintenance.storage.mapper;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.List;
+import java.util.ServiceLoader;
+import
org.apache.gravitino.maintenance.storage.mapper.provider.TmsMapperPackageProvider;
+import
org.apache.gravitino.storage.relational.mapper.provider.MapperPackageProvider;
+import org.junit.jupiter.api.Test;
+
+class TestTmsMapperPackageProvider {
+
+ @Test
+ void testMapperClasses() {
+ MapperPackageProvider provider = new TmsMapperPackageProvider();
+ assertEquals(2, provider.getMapperClasses().size());
+
assertTrue(provider.getMapperClasses().contains(TableMaintenanceActionMapper.class));
+
assertTrue(provider.getMapperClasses().contains(TableSnapshotMetricsMapper.class));
+ }
+
+ @Test
+ void testServiceLoaderRegistration() {
+ List<MapperPackageProvider> providers =
+ ServiceLoader.load(MapperPackageProvider.class).stream()
+ .map(ServiceLoader.Provider::get)
+ .filter(provider -> provider instanceof TmsMapperPackageProvider)
+ .toList();
+ assertEquals(1, providers.size());
+ assertTrue(providers.get(0) instanceof TmsMapperPackageProvider);
+ }
+}
diff --git
a/plugins/tms/src/test/java/org/apache/gravitino/maintenance/storage/service/TestTableMaintenanceActionMetaService.java
b/plugins/tms/src/test/java/org/apache/gravitino/maintenance/storage/service/TestTableMaintenanceActionMetaService.java
new file mode 100644
index 0000000000..2921255cd4
--- /dev/null
+++
b/plugins/tms/src/test/java/org/apache/gravitino/maintenance/storage/service/TestTableMaintenanceActionMetaService.java
@@ -0,0 +1,65 @@
+/*
+ * 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.maintenance.storage.service;
+
+import org.apache.gravitino.storage.relational.TestJDBCBackend;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.TestTemplate;
+
+class TestTableMaintenanceActionMetaService extends TestJDBCBackend {
+
+ @TestTemplate
+ void testInsertBindSampleFinishAndDeleteKeepsMetrics() {
+ TableMaintenanceActionMetaService actionService =
+ TableMaintenanceActionMetaService.getInstance();
+ TableSnapshotMetricsMetaService metricsService =
TableSnapshotMetricsMetaService.getInstance();
+
+ long before = System.currentTimeMillis();
+ long actionId = actionService.insertInFlight(10L, 20L);
+ long after = System.currentTimeMillis();
+ Assertions.assertNull(actionService.getById(actionId).jobId());
+ Assertions.assertTrue(actionService.getById(actionId).createdAt() >=
before);
+ Assertions.assertTrue(actionService.getById(actionId).createdAt() <=
after);
+
+ long jobId = 999L;
+ actionService.bindJobId(actionId, jobId);
+ Assertions.assertEquals(jobId, actionService.getByJobId(jobId).jobId());
+
+ actionService.recordBeforeSnapshot(jobId, 100L);
+ Assertions.assertEquals(100L,
actionService.getByJobId(jobId).beforeSnapshotId());
+ Assertions.assertThrows(
+ IllegalStateException.class, () ->
actionService.recordBeforeSnapshot(jobId, 101L));
+
+ metricsService.insertMetrics(10L, 20L, 100L, "{\"mse\":1}");
+ actionService.markFinished(jobId, 200L, System.currentTimeMillis());
+ Assertions.assertThrows(
+ IllegalStateException.class,
+ () -> actionService.markFinished(jobId, 201L,
System.currentTimeMillis()));
+ metricsService.insertMetrics(10L, 20L, 200L, "{\"mse\":2}");
+
+ actionService.deleteByJobId(jobId);
+
+ Assertions.assertNull(actionService.getByJobId(jobId));
+ Assertions.assertEquals(
+ "{\"mse\":1}", metricsService.getMetrics(10L, 20L,
100L).metricsValue());
+ Assertions.assertEquals(
+ "{\"mse\":2}", metricsService.getMetrics(10L, 20L,
200L).metricsValue());
+ }
+}
diff --git
a/plugins/tms/src/test/java/org/apache/gravitino/maintenance/storage/service/TestTableSnapshotMetricsMetaService.java
b/plugins/tms/src/test/java/org/apache/gravitino/maintenance/storage/service/TestTableSnapshotMetricsMetaService.java
new file mode 100644
index 0000000000..ca1fc00c79
--- /dev/null
+++
b/plugins/tms/src/test/java/org/apache/gravitino/maintenance/storage/service/TestTableSnapshotMetricsMetaService.java
@@ -0,0 +1,46 @@
+/*
+ * 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.maintenance.storage.service;
+
+import org.apache.gravitino.storage.relational.TestJDBCBackend;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.TestTemplate;
+
+class TestTableSnapshotMetricsMetaService extends TestJDBCBackend {
+
+ @TestTemplate
+ void testInsertAndGetMetrics() {
+ TableSnapshotMetricsMetaService service =
TableSnapshotMetricsMetaService.getInstance();
+ long before = System.currentTimeMillis();
+ service.insertMetrics(1L, 2L, 99L, "{\"mse\":1}");
+ long after = System.currentTimeMillis();
+ Assertions.assertEquals("{\"mse\":1}", service.getMetrics(1L, 2L,
99L).metricsValue());
+ Assertions.assertTrue(service.getMetrics(1L, 2L, 99L).createdAt() >=
before);
+ Assertions.assertTrue(service.getMetrics(1L, 2L, 99L).createdAt() <=
after);
+ }
+
+ @TestTemplate
+ void testInsertMetricsIsIdempotent() {
+ TableSnapshotMetricsMetaService service =
TableSnapshotMetricsMetaService.getInstance();
+ service.insertMetrics(3L, 4L, 88L, "{\"mse\":1}");
+ service.insertMetrics(3L, 4L, 88L, "{\"mse\":2}");
+ Assertions.assertEquals("{\"mse\":1}", service.getMetrics(3L, 4L,
88L).metricsValue());
+ }
+}
diff --git a/scripts/h2/schema-2.0.0-h2.sql b/scripts/h2/schema-2.0.0-h2.sql
index 373b8558bf..ac7fb308d1 100644
--- a/scripts/h2/schema-2.0.0-h2.sql
+++ b/scripts/h2/schema-2.0.0-h2.sql
@@ -680,3 +680,32 @@ CREATE TABLE IF NOT EXISTS `iceberg_cleanup_job` (
) 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`);
+
+CREATE TABLE IF NOT EXISTS `table_maintenance_action` (
+ `id` BIGINT AUTO_INCREMENT NOT NULL COMMENT 'auto increment id',
+ `job_id` BIGINT COMMENT 'optional job run id mapped to
job_run_meta.job_run_id, null until bound after runJob',
+ `table_id` BIGINT NOT NULL COMMENT 'table id from table_meta',
+ `policy_id` BIGINT NOT NULL COMMENT 'policy id from policy_meta',
+ `before_snapshot_id` BIGINT COMMENT 'Iceberg snapshot id before
maintenance, null until sampled',
+ `after_snapshot_id` BIGINT COMMENT 'Iceberg snapshot id after maintenance,
null while pending',
+ `finished_at` BIGINT COMMENT 'finish timestamp in milliseconds, null while
in flight',
+ `created_at` BIGINT NOT NULL COMMENT 'creation timestamp in milliseconds',
+ PRIMARY KEY (`id`)
+) COMMENT='table maintenance action occupancy and run metadata';
+CREATE UNIQUE INDEX IF NOT EXISTS `uk_tma_job_id` ON
`table_maintenance_action`(`job_id`);
+CREATE INDEX IF NOT EXISTS `idx_tma_table_policy_finished` ON
`table_maintenance_action`(`table_id`, `policy_id`, `finished_at`);
+CREATE INDEX IF NOT EXISTS `idx_tma_before_sid` ON
`table_maintenance_action`(`table_id`, `before_snapshot_id`);
+CREATE INDEX IF NOT EXISTS `idx_tma_after_sid` ON
`table_maintenance_action`(`table_id`, `after_snapshot_id`);
+CREATE INDEX IF NOT EXISTS `idx_tma_created_at` ON
`table_maintenance_action`(`created_at`);
+
+CREATE TABLE IF NOT EXISTS `table_snapshot_metrics` (
+ `id` BIGINT AUTO_INCREMENT NOT NULL COMMENT 'auto increment id',
+ `table_id` BIGINT NOT NULL COMMENT 'table id from table_meta',
+ `policy_id` BIGINT NOT NULL COMMENT 'policy id from policy_meta',
+ `snapshot_id` BIGINT NOT NULL COMMENT 'Iceberg snapshot id',
+ `metrics_value` CLOB NOT NULL COMMENT 'metrics value as JSON',
+ `created_at` BIGINT NOT NULL COMMENT 'creation timestamp in milliseconds',
+ PRIMARY KEY (`id`)
+) COMMENT='table snapshot metrics metadata';
+CREATE UNIQUE INDEX IF NOT EXISTS `uk_tid_pid_sid` ON
`table_snapshot_metrics`(`table_id`, `policy_id`, `snapshot_id`);
+CREATE INDEX IF NOT EXISTS `idx_tsm_created_at` ON
`table_snapshot_metrics`(`created_at`);
diff --git a/scripts/h2/upgrade-1.3.0-to-2.0.0-h2.sql
b/scripts/h2/upgrade-1.3.0-to-2.0.0-h2.sql
index 6b66d9e791..fb4f838a66 100644
--- a/scripts/h2/upgrade-1.3.0-to-2.0.0-h2.sql
+++ b/scripts/h2/upgrade-1.3.0-to-2.0.0-h2.sql
@@ -102,6 +102,35 @@ CREATE TABLE IF NOT EXISTS `semantic_model_version_info` (
KEY `idx_smvi_sid` (`schema_id`)
) ENGINE=InnoDB COMMENT 'semantic model version information';
+CREATE TABLE IF NOT EXISTS `table_maintenance_action` (
+ `id` BIGINT AUTO_INCREMENT NOT NULL COMMENT 'auto increment id',
+ `job_id` BIGINT COMMENT 'optional job run id mapped to
job_run_meta.job_run_id, null until bound after runJob',
+ `table_id` BIGINT NOT NULL COMMENT 'table id from table_meta',
+ `policy_id` BIGINT NOT NULL COMMENT 'policy id from policy_meta',
+ `before_snapshot_id` BIGINT COMMENT 'Iceberg snapshot id before
maintenance, null until sampled',
+ `after_snapshot_id` BIGINT COMMENT 'Iceberg snapshot id after maintenance,
null while pending',
+ `finished_at` BIGINT COMMENT 'finish timestamp in milliseconds, null while
in flight',
+ `created_at` BIGINT NOT NULL COMMENT 'creation timestamp in milliseconds',
+ PRIMARY KEY (`id`)
+) COMMENT='table maintenance action occupancy and run metadata';
+CREATE UNIQUE INDEX IF NOT EXISTS `uk_tma_job_id` ON
`table_maintenance_action`(`job_id`);
+CREATE INDEX IF NOT EXISTS `idx_tma_table_policy_finished` ON
`table_maintenance_action`(`table_id`, `policy_id`, `finished_at`);
+CREATE INDEX IF NOT EXISTS `idx_tma_before_sid` ON
`table_maintenance_action`(`table_id`, `before_snapshot_id`);
+CREATE INDEX IF NOT EXISTS `idx_tma_after_sid` ON
`table_maintenance_action`(`table_id`, `after_snapshot_id`);
+CREATE INDEX IF NOT EXISTS `idx_tma_created_at` ON
`table_maintenance_action`(`created_at`);
+
+CREATE TABLE IF NOT EXISTS `table_snapshot_metrics` (
+ `id` BIGINT AUTO_INCREMENT NOT NULL COMMENT 'auto increment id',
+ `table_id` BIGINT NOT NULL COMMENT 'table id from table_meta',
+ `policy_id` BIGINT NOT NULL COMMENT 'policy id from policy_meta',
+ `snapshot_id` BIGINT NOT NULL COMMENT 'Iceberg snapshot id',
+ `metrics_value` CLOB NOT NULL COMMENT 'metrics value as JSON',
+ `created_at` BIGINT NOT NULL COMMENT 'creation timestamp in milliseconds',
+ PRIMARY KEY (`id`)
+) COMMENT='table snapshot metrics metadata';
+CREATE UNIQUE INDEX IF NOT EXISTS `uk_tid_pid_sid` ON
`table_snapshot_metrics`(`table_id`, `policy_id`, `snapshot_id`);
+CREATE INDEX IF NOT EXISTS `idx_tsm_created_at` ON
`table_snapshot_metrics`(`created_at`);
+
-- Merge duplicate live owners left by concurrent assignments: the newest live
row
-- (largest id) wins, and older ones are soft-deleted.
UPDATE `owner_meta`
diff --git a/scripts/mysql/schema-2.0.0-mysql.sql
b/scripts/mysql/schema-2.0.0-mysql.sql
index 0edbb0f1f2..786356bcf1 100644
--- a/scripts/mysql/schema-2.0.0-mysql.sql
+++ b/scripts/mysql/schema-2.0.0-mysql.sql
@@ -667,3 +667,34 @@ CREATE TABLE IF NOT EXISTS `iceberg_cleanup_job` (
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';
+
+CREATE TABLE IF NOT EXISTS `table_maintenance_action` (
+ `id` BIGINT(20) UNSIGNED NOT NULL AUTO_INCREMENT COMMENT 'auto increment
id',
+ `job_id` BIGINT(20) UNSIGNED NULL COMMENT 'optional job run id mapped to
job_run_meta.job_run_id, null until bound after runJob',
+ `table_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'table id from table_meta',
+ `policy_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'policy id from
policy_meta',
+ `before_snapshot_id` BIGINT(20) NULL COMMENT 'Iceberg snapshot id before
maintenance, null until sampled',
+ `after_snapshot_id` BIGINT(20) NULL COMMENT 'Iceberg snapshot id after
maintenance, null while pending',
+ `finished_at` BIGINT(20) UNSIGNED NULL COMMENT 'finish timestamp in
milliseconds, null while in flight',
+ `created_at` BIGINT(20) UNSIGNED NOT NULL COMMENT 'creation timestamp in
milliseconds',
+ PRIMARY KEY (`id`),
+ UNIQUE KEY `uk_tma_job_id` (`job_id`),
+ KEY `idx_tma_table_policy_finished` (`table_id`, `policy_id`,
`finished_at`),
+ KEY `idx_tma_before_sid` (`table_id`, `before_snapshot_id`),
+ KEY `idx_tma_after_sid` (`table_id`, `after_snapshot_id`),
+ KEY `idx_tma_created_at` (`created_at`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin
+ COMMENT 'table maintenance action occupancy and run metadata';
+
+CREATE TABLE IF NOT EXISTS `table_snapshot_metrics` (
+ `id` BIGINT(20) UNSIGNED NOT NULL AUTO_INCREMENT COMMENT 'auto increment
id',
+ `table_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'table id from table_meta',
+ `policy_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'policy id from
policy_meta',
+ `snapshot_id` BIGINT(20) NOT NULL COMMENT 'Iceberg snapshot id',
+ `metrics_value` MEDIUMTEXT NOT NULL COMMENT 'metrics value as JSON',
+ `created_at` BIGINT(20) UNSIGNED NOT NULL COMMENT 'creation timestamp in
milliseconds',
+ PRIMARY KEY (`id`),
+ UNIQUE KEY `uk_tid_pid_sid` (`table_id`, `policy_id`, `snapshot_id`),
+ KEY `idx_tsm_created_at` (`created_at`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin
+ COMMENT 'table snapshot metrics metadata';
diff --git a/scripts/mysql/upgrade-1.3.0-to-2.0.0-mysql.sql
b/scripts/mysql/upgrade-1.3.0-to-2.0.0-mysql.sql
index 2815471571..c5108a79c9 100644
--- a/scripts/mysql/upgrade-1.3.0-to-2.0.0-mysql.sql
+++ b/scripts/mysql/upgrade-1.3.0-to-2.0.0-mysql.sql
@@ -549,6 +549,37 @@ CREATE TABLE IF NOT EXISTS `semantic_model_version_info` (
KEY `idx_smvi_sid` (`schema_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT 'semantic
model version information';
+CREATE TABLE IF NOT EXISTS `table_maintenance_action` (
+ `id` BIGINT(20) UNSIGNED NOT NULL AUTO_INCREMENT COMMENT 'auto increment
id',
+ `job_id` BIGINT(20) UNSIGNED NULL COMMENT 'optional job run id mapped to
job_run_meta.job_run_id, null until bound after runJob',
+ `table_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'table id from table_meta',
+ `policy_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'policy id from
policy_meta',
+ `before_snapshot_id` BIGINT(20) NULL COMMENT 'Iceberg snapshot id before
maintenance, null until sampled',
+ `after_snapshot_id` BIGINT(20) NULL COMMENT 'Iceberg snapshot id after
maintenance, null while pending',
+ `finished_at` BIGINT(20) UNSIGNED NULL COMMENT 'finish timestamp in
milliseconds, null while in flight',
+ `created_at` BIGINT(20) UNSIGNED NOT NULL COMMENT 'creation timestamp in
milliseconds',
+ PRIMARY KEY (`id`),
+ UNIQUE KEY `uk_tma_job_id` (`job_id`),
+ KEY `idx_tma_table_policy_finished` (`table_id`, `policy_id`,
`finished_at`),
+ KEY `idx_tma_before_sid` (`table_id`, `before_snapshot_id`),
+ KEY `idx_tma_after_sid` (`table_id`, `after_snapshot_id`),
+ KEY `idx_tma_created_at` (`created_at`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin
+ COMMENT 'table maintenance action occupancy and run metadata';
+
+CREATE TABLE IF NOT EXISTS `table_snapshot_metrics` (
+ `id` BIGINT(20) UNSIGNED NOT NULL AUTO_INCREMENT COMMENT 'auto increment
id',
+ `table_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'table id from table_meta',
+ `policy_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'policy id from
policy_meta',
+ `snapshot_id` BIGINT(20) NOT NULL COMMENT 'Iceberg snapshot id',
+ `metrics_value` MEDIUMTEXT NOT NULL COMMENT 'metrics value as JSON',
+ `created_at` BIGINT(20) UNSIGNED NOT NULL COMMENT 'creation timestamp in
milliseconds',
+ PRIMARY KEY (`id`),
+ UNIQUE KEY `uk_tid_pid_sid` (`table_id`, `policy_id`, `snapshot_id`),
+ KEY `idx_tsm_created_at` (`created_at`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin
+ COMMENT 'table snapshot metrics metadata';
+
-- Merge duplicate live owners left by concurrent assignments: the newest live
row
-- (largest id) wins, and older ones are soft-deleted.
UPDATE `owner_meta` o
diff --git a/scripts/postgresql/schema-2.0.0-postgresql.sql
b/scripts/postgresql/schema-2.0.0-postgresql.sql
index 3fe920d0eb..b64a8d7fbb 100644
--- a/scripts/postgresql/schema-2.0.0-postgresql.sql
+++ b/scripts/postgresql/schema-2.0.0-postgresql.sql
@@ -1175,3 +1175,46 @@ COMMENT ON COLUMN iceberg_cleanup_job.last_error IS
'truncated reason for the mo
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';
+
+CREATE TABLE IF NOT EXISTS table_maintenance_action (
+ id BIGSERIAL PRIMARY KEY,
+ job_id BIGINT,
+ table_id BIGINT NOT NULL,
+ policy_id BIGINT NOT NULL,
+ before_snapshot_id BIGINT,
+ after_snapshot_id BIGINT,
+ finished_at BIGINT,
+ created_at BIGINT NOT NULL
+);
+CREATE UNIQUE INDEX IF NOT EXISTS uk_tma_job_id ON table_maintenance_action
(job_id);
+CREATE INDEX IF NOT EXISTS idx_tma_table_policy_finished ON
table_maintenance_action (table_id, policy_id, finished_at);
+CREATE INDEX IF NOT EXISTS idx_tma_before_sid ON table_maintenance_action
(table_id, before_snapshot_id);
+CREATE INDEX IF NOT EXISTS idx_tma_after_sid ON table_maintenance_action
(table_id, after_snapshot_id);
+CREATE INDEX IF NOT EXISTS idx_tma_created_at ON table_maintenance_action
(created_at);
+COMMENT ON TABLE table_maintenance_action IS 'table maintenance action
occupancy and run metadata';
+COMMENT ON COLUMN table_maintenance_action.id IS 'auto increment id';
+COMMENT ON COLUMN table_maintenance_action.job_id IS 'optional job run id
mapped to job_run_meta.job_run_id, null until bound after runJob';
+COMMENT ON COLUMN table_maintenance_action.table_id IS 'table id from
table_meta';
+COMMENT ON COLUMN table_maintenance_action.policy_id IS 'policy id from
policy_meta';
+COMMENT ON COLUMN table_maintenance_action.before_snapshot_id IS 'Iceberg
snapshot id before maintenance, null until sampled';
+COMMENT ON COLUMN table_maintenance_action.after_snapshot_id IS 'Iceberg
snapshot id after maintenance, null while pending';
+COMMENT ON COLUMN table_maintenance_action.finished_at IS 'finish timestamp in
milliseconds, null while in flight';
+COMMENT ON COLUMN table_maintenance_action.created_at IS 'creation timestamp
in milliseconds';
+
+CREATE TABLE IF NOT EXISTS table_snapshot_metrics (
+ id BIGSERIAL PRIMARY KEY,
+ table_id BIGINT NOT NULL,
+ policy_id BIGINT NOT NULL,
+ snapshot_id BIGINT NOT NULL,
+ metrics_value TEXT NOT NULL,
+ created_at BIGINT NOT NULL
+);
+CREATE UNIQUE INDEX IF NOT EXISTS uk_tid_pid_sid ON table_snapshot_metrics
(table_id, policy_id, snapshot_id);
+CREATE INDEX IF NOT EXISTS idx_tsm_created_at ON table_snapshot_metrics
(created_at);
+COMMENT ON TABLE table_snapshot_metrics IS 'table snapshot metrics metadata';
+COMMENT ON COLUMN table_snapshot_metrics.id IS 'auto increment id';
+COMMENT ON COLUMN table_snapshot_metrics.table_id IS 'table id from
table_meta';
+COMMENT ON COLUMN table_snapshot_metrics.policy_id IS 'policy id from
policy_meta';
+COMMENT ON COLUMN table_snapshot_metrics.snapshot_id IS 'Iceberg snapshot id';
+COMMENT ON COLUMN table_snapshot_metrics.metrics_value IS 'metrics value as
JSON';
+COMMENT ON COLUMN table_snapshot_metrics.created_at IS 'creation timestamp in
milliseconds';
diff --git a/scripts/postgresql/upgrade-1.3.0-to-2.0.0-postgresql.sql
b/scripts/postgresql/upgrade-1.3.0-to-2.0.0-postgresql.sql
index 771b9ccd67..6fde2a8e27 100644
--- a/scripts/postgresql/upgrade-1.3.0-to-2.0.0-postgresql.sql
+++ b/scripts/postgresql/upgrade-1.3.0-to-2.0.0-postgresql.sql
@@ -151,6 +151,49 @@ COMMENT ON COLUMN semantic_model_version_info.properties
IS 'semantic model prop
COMMENT ON COLUMN semantic_model_version_info.audit_info IS 'semantic model
version audit info';
COMMENT ON COLUMN semantic_model_version_info.deleted_at IS 'version deleted
at';
+CREATE TABLE IF NOT EXISTS table_maintenance_action (
+ id BIGSERIAL PRIMARY KEY,
+ job_id BIGINT,
+ table_id BIGINT NOT NULL,
+ policy_id BIGINT NOT NULL,
+ before_snapshot_id BIGINT,
+ after_snapshot_id BIGINT,
+ finished_at BIGINT,
+ created_at BIGINT NOT NULL
+);
+CREATE UNIQUE INDEX IF NOT EXISTS uk_tma_job_id ON table_maintenance_action
(job_id);
+CREATE INDEX IF NOT EXISTS idx_tma_table_policy_finished ON
table_maintenance_action (table_id, policy_id, finished_at);
+CREATE INDEX IF NOT EXISTS idx_tma_before_sid ON table_maintenance_action
(table_id, before_snapshot_id);
+CREATE INDEX IF NOT EXISTS idx_tma_after_sid ON table_maintenance_action
(table_id, after_snapshot_id);
+CREATE INDEX IF NOT EXISTS idx_tma_created_at ON table_maintenance_action
(created_at);
+COMMENT ON TABLE table_maintenance_action IS 'table maintenance action
occupancy and run metadata';
+COMMENT ON COLUMN table_maintenance_action.id IS 'auto increment id';
+COMMENT ON COLUMN table_maintenance_action.job_id IS 'optional job run id
mapped to job_run_meta.job_run_id, null until bound after runJob';
+COMMENT ON COLUMN table_maintenance_action.table_id IS 'table id from
table_meta';
+COMMENT ON COLUMN table_maintenance_action.policy_id IS 'policy id from
policy_meta';
+COMMENT ON COLUMN table_maintenance_action.before_snapshot_id IS 'Iceberg
snapshot id before maintenance, null until sampled';
+COMMENT ON COLUMN table_maintenance_action.after_snapshot_id IS 'Iceberg
snapshot id after maintenance, null while pending';
+COMMENT ON COLUMN table_maintenance_action.finished_at IS 'finish timestamp in
milliseconds, null while in flight';
+COMMENT ON COLUMN table_maintenance_action.created_at IS 'creation timestamp
in milliseconds';
+
+CREATE TABLE IF NOT EXISTS table_snapshot_metrics (
+ id BIGSERIAL PRIMARY KEY,
+ table_id BIGINT NOT NULL,
+ policy_id BIGINT NOT NULL,
+ snapshot_id BIGINT NOT NULL,
+ metrics_value TEXT NOT NULL,
+ created_at BIGINT NOT NULL
+);
+CREATE UNIQUE INDEX IF NOT EXISTS uk_tid_pid_sid ON table_snapshot_metrics
(table_id, policy_id, snapshot_id);
+CREATE INDEX IF NOT EXISTS idx_tsm_created_at ON table_snapshot_metrics
(created_at);
+COMMENT ON TABLE table_snapshot_metrics IS 'table snapshot metrics metadata';
+COMMENT ON COLUMN table_snapshot_metrics.id IS 'auto increment id';
+COMMENT ON COLUMN table_snapshot_metrics.table_id IS 'table id from
table_meta';
+COMMENT ON COLUMN table_snapshot_metrics.policy_id IS 'policy id from
policy_meta';
+COMMENT ON COLUMN table_snapshot_metrics.snapshot_id IS 'Iceberg snapshot id';
+COMMENT ON COLUMN table_snapshot_metrics.metrics_value IS 'metrics value as
JSON';
+COMMENT ON COLUMN table_snapshot_metrics.created_at IS 'creation timestamp in
milliseconds';
+
-- Merge duplicate live owners left by concurrent assignments: the newest live
row
-- (largest id) wins, and older ones are soft-deleted.
UPDATE owner_meta
diff --git a/settings.gradle.kts b/settings.gradle.kts
index a8ec768322..ced9feab65 100644
--- a/settings.gradle.kts
+++ b/settings.gradle.kts
@@ -127,6 +127,7 @@ include(":catalogs:hadoop-auth")
include(":lineage")
include(":mcp-server")
include(":plugins:idp-basic")
+include(":plugins:tms")
include(
":maintenance:optimizer-api",
":maintenance:updaters",
