This is an automated email from the ASF dual-hosted git repository.
roryqi pushed a commit to branch branch-tms-scheduler
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/branch-tms-scheduler by this
push:
new 70d5e9bc48 [#14065] feat(tms): In-job recommenders and terminal
occupancy reclaim (#14054)
70d5e9bc48 is described below
commit 70d5e9bc4801a50576093566f937ec18f78229b2
Author: MaSai <[email protected]>
AuthorDate: Thu Oct 22 09:08:42 2026 +0800
[#14065] feat(tms): In-job recommenders and terminal occupancy reclaim
(#14054)
### What changes were proposed in this pull request?
First slice of the in-job track. Targets `branch-tms-scheduler`.
- `TypeDispatchingInJobRecommender` routes by policy type. Compaction
uses `CompactionTriggerEvaluator`. Orphan cleanup evaluates the policy
`trigger-expr`. Expire and rewrite-manifests placeholders use
`AlwaysMaintainInJobRecommender`.
- Recommenders and `OccupancyGate` keep `PolicyMetaService` and
`JobMetaService` in static fields and call those services. TMS does not
call core mappers.
- `PolicyMetaService.getPolicyById` loads one active policy, its current
version, and the metalake name in one join.
- `OccupancyGate.reclaimTerminalBound` sets `finished_at` when the bound
job is already terminal, so the soft slot can be claimed again.
- `ExpressionEvaluator` and `QLExpressionEvaluator` live in
`optimizer-api`, with `CompactionTriggerEvaluator` beside them.
### Why are the changes needed?
In-job sample/finish needs a maintain or skip decision, and a bound
action must release its soft slot after the job is already terminal.
Policy content is loaded by id through `PolicyMetaService`.
Part of #13760.
Fix: #14065
### Does this PR introduce _any_ user-facing change?
No public REST. Occupancy reclaim changes for a maintenance action whose
bound job is already terminal.
### How was this patch tested?
`./gradlew :plugins:tms:test -PskipITs -PskipDockerTests=true --tests
org.apache.gravitino.maintenance.injob.* --tests
org.apache.gravitino.maintenance.occupancy.TestOccupancyGate`
`./gradlew :core:coreH2Test -PskipITs --tests
org.apache.gravitino.storage.relational.service.TestPolicyMetaService.testGetPolicyById`
---------
Co-authored-by: Cursor <[email protected]>
---
.../relational/mapper/PolicyMetaMapper.java | 31 ++++
.../mapper/PolicyMetaSQLProviderFactory.java | 4 +
.../provider/base/PolicyMetaBaseSQLProvider.java | 27 ++++
.../gravitino/storage/relational/po/PolicyPO.java | 3 +
.../relational/service/PolicyMetaService.java | 24 +++
.../relational/service/TableMetaService.java | 35 +++++
.../base/TestPolicyMetaBaseSQLProvider.java | 39 +++++
.../relational/service/TestPolicyMetaService.java | 22 +++
.../relational/service/TestTableMetaService.java | 18 +++
maintenance/optimizer-api/build.gradle.kts | 3 +
.../compaction/CompactionTriggerEvaluator.java | 76 +++++++++
.../recommender/util/ExpressionEvaluator.java | 0
.../recommender/util/QLExpressionEvaluator.java | 4 +
plugins/tms/build.gradle.kts | 3 +
...er.java => AlwaysMaintainInJobRecommender.java} | 30 ++--
.../injob/CompactionInJobRecommender.java | 122 +++++++++++++++
.../maintenance/injob/InJobRecommender.java | 29 +++-
.../gravitino/maintenance/injob/MetricsJson.java | 70 +++++++++
.../maintenance/injob/RecommendDecision.java | 5 +-
.../injob/TriggerExprInJobRecommender.java | 128 ++++++++++++++++
.../injob/TypeDispatchingInJobRecommender.java | 138 +++++++++++++++++
.../maintenance/occupancy/OccupancyGate.java | 162 +++++++++++++++-----
.../mapper/TableMaintenanceActionMapper.java | 6 +
.../TableMaintenanceActionSQLProviderFactory.java | 10 ++
.../TableMaintenanceActionBaseSQLProvider.java | 16 ++
.../service/TableMaintenanceActionMetaService.java | 25 ++-
.../injob/TestCompactionInJobRecommender.java | 64 ++++++++
.../maintenance/injob/TestMetricsJson.java | 44 ++++++
.../injob/TestTriggerExprInJobRecommender.java | 65 ++++++++
.../injob/TestTypeDispatchingInJobRecommender.java | 146 ++++++++++++++++++
.../maintenance/occupancy/TestOccupancyGate.java | 169 +++++++++++++++++++--
.../TestTableMaintenanceActionMetaService.java | 26 ++++
32 files changed, 1474 insertions(+), 70 deletions(-)
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyMetaMapper.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyMetaMapper.java
index 383791342b..ecd4b356a3 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyMetaMapper.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyMetaMapper.java
@@ -86,6 +86,37 @@ public interface PolicyMetaMapper {
PolicyPO selectPolicyMetaByMetalakeAndName(
@Param("metalakeName") String metalakeName, @Param("policyName") String
policyName);
+ /**
+ * Loads one active policy, its current version, and the owning metalake
name.
+ *
+ * @param policyId policy id
+ * @return the policy row, or null when the policy or its metalake is not
active
+ */
+ @Results(
+ id = "policyWithMetalakeNameResultMap",
+ value = {
+ @Result(property = "policyId", column = "policy_id"),
+ @Result(property = "policyName", column = "policy_name"),
+ @Result(property = "policyType", column = "policy_type"),
+ @Result(property = "metalakeId", column = "metalake_id"),
+ @Result(property = "metalakeName", column = "metalake_name"),
+ @Result(property = "auditInfo", column = "audit_info"),
+ @Result(property = "currentVersion", column = "current_version"),
+ @Result(property = "lastVersion", column = "last_version"),
+ @Result(property = "occVersion", column = "occ_version"),
+ @Result(property = "deletedAt", column = "deleted_at"),
+ @Result(property = "policyVersionPO.id", column = "id"),
+ @Result(property = "policyVersionPO.metalakeId", column =
"version_metalake_id"),
+ @Result(property = "policyVersionPO.policyId", column =
"version_policy_id"),
+ @Result(property = "policyVersionPO.version", column = "version"),
+ @Result(property = "policyVersionPO.policyComment", column =
"policy_comment"),
+ @Result(property = "policyVersionPO.enabled", column = "enabled"),
+ @Result(property = "policyVersionPO.content", column = "content"),
+ @Result(property = "policyVersionPO.deletedAt", column =
"version_deleted_at")
+ })
+ @SelectProvider(type = PolicyMetaSQLProviderFactory.class, method =
"selectPolicyMetaByPolicyId")
+ PolicyPO selectPolicyMetaByPolicyId(@Param("policyId") Long policyId);
+
@UpdateProvider(type = PolicyMetaSQLProviderFactory.class, method =
"updatePolicyMeta")
Integer updatePolicyMeta(
@Param("newPolicyMeta") PolicyPO newPolicyMeta,
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyMetaSQLProviderFactory.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyMetaSQLProviderFactory.java
index 602d789c8e..1b3d53b153 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyMetaSQLProviderFactory.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/PolicyMetaSQLProviderFactory.java
@@ -98,6 +98,10 @@ public class PolicyMetaSQLProviderFactory {
return
getProvider().selectPolicyMetaByMetalakeIdAndNameForUpdate(metalakeId,
policyName);
}
+ public static String selectPolicyMetaByPolicyId(@Param("policyId") Long
policyId) {
+ return getProvider().selectPolicyMetaByPolicyId(policyId);
+ }
+
public static String selectPolicyByPolicyId(@Param("policyId") Long
policyId) {
return getProvider().selectPolicyByPolicyId(policyId);
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/PolicyMetaBaseSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/PolicyMetaBaseSQLProvider.java
index 94993c6f17..c0fb2227ef 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/PolicyMetaBaseSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/PolicyMetaBaseSQLProvider.java
@@ -149,6 +149,33 @@ public class PolicyMetaBaseSQLProvider {
+ " WHERE metalake_id = #{metalakeId} AND deleted_at = 0";
}
+ /**
+ * One join of policy, metalake, and the current policy version, selected by
policy id.
+ *
+ * @param policyId policy id
+ * @return the select statement
+ */
+ public String selectPolicyMetaByPolicyId(@Param("policyId") Long policyId) {
+ return "SELECT pm.policy_id, pm.policy_name, pm.policy_type,
pm.metalake_id,"
+ + " pm.audit_info, pm.current_version, pm.last_version,
pm.occ_version,"
+ + " pm.deleted_at, mm.metalake_name as metalake_name,"
+ + " pvi.id, pvi.metalake_id as version_metalake_id, pvi.policy_id as
version_policy_id,"
+ + " pvi.version, pvi.policy_comment, pvi.enabled, pvi.content,"
+ + " pvi.deleted_at as version_deleted_at"
+ + " FROM "
+ + POLICY_META_TABLE_NAME
+ + " pm JOIN "
+ + MetalakeMetaMapper.TABLE_NAME
+ + " mm ON pm.metalake_id = mm.metalake_id"
+ + " JOIN "
+ + POLICY_VERSION_TABLE_NAME
+ + " pvi ON pm.policy_id = pvi.policy_id"
+ + " AND pm.current_version = pvi.version"
+ + " WHERE pm.policy_id = #{policyId}"
+ + " AND pm.deleted_at = 0 AND mm.deleted_at = 0"
+ + " AND pvi.deleted_at = 0";
+ }
+
public String selectPolicyByPolicyId(@Param("policyId") Long policyId) {
return "SELECT pm.policy_id, pm.policy_name, pm.policy_type,
pm.metalake_id,"
+ " pm.audit_info, pm.current_version, pm.last_version,
pm.occ_version,"
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/po/PolicyPO.java
b/core/src/main/java/org/apache/gravitino/storage/relational/po/PolicyPO.java
index 76e0fb4f9a..04893d833e 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/po/PolicyPO.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/po/PolicyPO.java
@@ -28,6 +28,9 @@ public class PolicyPO {
private String policyName;
private String policyType;
private Long metalakeId;
+ /** Metalake name from a join. Absent on queries that do not select it. */
+ private String metalakeName;
+
private String auditInfo;
private Long currentVersion;
private Long lastVersion;
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/PolicyMetaService.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/PolicyMetaService.java
index 2f30649b0b..4d3a1abc39 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/service/PolicyMetaService.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/PolicyMetaService.java
@@ -30,6 +30,7 @@ import java.util.Map;
import java.util.Objects;
import java.util.function.Function;
import java.util.stream.Collectors;
+import javax.annotation.Nullable;
import org.apache.gravitino.Entity;
import org.apache.gravitino.EntityAlreadyExistsException;
import org.apache.gravitino.HasIdentifier;
@@ -52,6 +53,7 @@ import org.apache.gravitino.storage.relational.po.PolicyPO;
import org.apache.gravitino.storage.relational.utils.ExceptionUtils;
import org.apache.gravitino.storage.relational.utils.POConverters;
import org.apache.gravitino.storage.relational.utils.SessionUtils;
+import org.apache.gravitino.utils.NamespaceUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -87,6 +89,28 @@ public class PolicyMetaService {
return POConverters.fromPolicyPO(policyPO, ident.namespace());
}
+ /**
+ * Loads one active policy by id, including its current version and metalake
name.
+ *
+ * <p>The lookup is a single join across policy, metalake, and the current
policy version.
+ *
+ * @param policyId policy id
+ * @return the policy, or {@code null} when the policy or its metalake is
not active
+ */
+ @Nullable
+ @Monitored(
+ metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
+ baseMetricName = "getPolicyById")
+ public PolicyEntity getPolicyById(long policyId) {
+ PolicyPO policyPO =
+ SessionUtils.getWithoutCommit(
+ PolicyMetaMapper.class, mapper ->
mapper.selectPolicyMetaByPolicyId(policyId));
+ if (policyPO == null || policyPO.getMetalakeName() == null) {
+ return null;
+ }
+ return POConverters.fromPolicyPO(policyPO,
NamespaceUtil.ofPolicy(policyPO.getMetalakeName()));
+ }
+
@Monitored(
metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
baseMetricName = "insertPolicy")
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/TableMetaService.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/TableMetaService.java
index 62a1d0a4c9..6f405747c6 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/service/TableMetaService.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/TableMetaService.java
@@ -23,10 +23,12 @@ import static
org.apache.gravitino.metrics.source.MetricsSource.GRAVITINO_RELATI
import com.google.common.base.Preconditions;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
+import javax.annotation.Nullable;
import org.apache.gravitino.Entity;
import org.apache.gravitino.HasIdentifier;
import org.apache.gravitino.MetadataObject;
@@ -37,6 +39,7 @@ import org.apache.gravitino.meta.ColumnEntity;
import org.apache.gravitino.meta.NamespacedEntityId;
import org.apache.gravitino.meta.TableEntity;
import org.apache.gravitino.metrics.Monitored;
+import org.apache.gravitino.storage.relational.mapper.MetalakeMetaMapper;
import org.apache.gravitino.storage.relational.mapper.OwnerMetaMapper;
import org.apache.gravitino.storage.relational.mapper.SecurableObjectMapper;
import org.apache.gravitino.storage.relational.mapper.StatisticMetaMapper;
@@ -44,6 +47,7 @@ import
org.apache.gravitino.storage.relational.mapper.TableMetaMapper;
import org.apache.gravitino.storage.relational.mapper.TableVersionMapper;
import
org.apache.gravitino.storage.relational.mapper.TagMetadataObjectRelMapper;
import org.apache.gravitino.storage.relational.po.ColumnPO;
+import org.apache.gravitino.storage.relational.po.MetalakePO;
import org.apache.gravitino.storage.relational.po.TablePO;
import org.apache.gravitino.storage.relational.utils.ExceptionUtils;
import org.apache.gravitino.storage.relational.utils.POConverters;
@@ -85,6 +89,37 @@ public class TableMetaService {
return tablePO.getTableId();
}
+ /**
+ * Returns the metalake name that owns an active table id.
+ *
+ * <p>Used when callers have a table entity id but not a metalake name (for
example TMS occupancy
+ * reclaim looking up a bound job without depending on the current policy
still existing).
+ *
+ * @param tableId table entity id
+ * @return the metalake name, or {@code null} when the table or metalake is
not active
+ */
+ @Nullable
+ @Monitored(
+ metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
+ baseMetricName = "getMetalakeNameByTableId")
+ public String getMetalakeNameByTableId(long tableId) {
+ List<TablePO> tables =
+ SessionUtils.getWithoutCommit(
+ TableMetaMapper.class,
+ mapper ->
mapper.listTablePOsByTableIds(Collections.singletonList(tableId)));
+ if (tables == null || tables.isEmpty()) {
+ return null;
+ }
+ Long metalakeId = tables.get(0).getMetalakeId();
+ if (metalakeId == null) {
+ return null;
+ }
+ MetalakePO metalake =
+ SessionUtils.getWithoutCommit(
+ MetalakeMetaMapper.class, mapper ->
mapper.selectMetalakeMetaById(metalakeId));
+ return metalake == null ? null : metalake.getMetalakeName();
+ }
+
@Monitored(
metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
baseMetricName = "getTableByIdentifier")
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/base/TestPolicyMetaBaseSQLProvider.java
b/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/base/TestPolicyMetaBaseSQLProvider.java
new file mode 100644
index 0000000000..e8d0c647b2
--- /dev/null
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/base/TestPolicyMetaBaseSQLProvider.java
@@ -0,0 +1,39 @@
+/*
+ * 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.storage.relational.mapper.provider.base;
+
+import org.apache.gravitino.storage.relational.mapper.MetalakeMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.PolicyVersionMapper;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class TestPolicyMetaBaseSQLProvider {
+
+ @Test
+ void testSelectPolicyMetaByPolicyIdIsOneJoin() {
+ String sql = new
PolicyMetaBaseSQLProvider().selectPolicyMetaByPolicyId(1L);
+
+ Assertions.assertEquals(sql.indexOf("SELECT"), sql.lastIndexOf("SELECT"));
+ Assertions.assertFalse(sql.contains(";"));
+ Assertions.assertTrue(sql.contains("JOIN " +
MetalakeMetaMapper.TABLE_NAME));
+ Assertions.assertTrue(sql.contains("JOIN " +
PolicyVersionMapper.POLICY_VERSION_TABLE_NAME));
+ Assertions.assertTrue(sql.contains("pm.policy_id = #{policyId}"));
+ Assertions.assertTrue(sql.contains("mm.metalake_name as metalake_name"));
+ }
+}
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestPolicyMetaService.java
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestPolicyMetaService.java
index 06b44444f6..74dc33e67b 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestPolicyMetaService.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestPolicyMetaService.java
@@ -240,6 +240,28 @@ public class TestPolicyMetaService extends TestJDBCBackend
{
assertEquals(1, versionDeletedMap2.values().stream().filter(value -> value
== 0L).count());
}
+ @TestTemplate
+ public void testGetPolicyById() throws IOException {
+ BaseMetalake metalake = createAndInsertMakeLake(METALAKE_NAME);
+ PolicyMetaService policyMetaService = PolicyMetaService.getInstance();
+ Assertions.assertNull(policyMetaService.getPolicyById(404L));
+
+ PolicyEntity policyEntity =
+ PolicyEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName("policy-by-id")
+ .withNamespace(NamespaceUtil.ofPolicy(metalake.name()))
+ .withComment("comment")
+ .withPolicyType(Policy.BuiltInType.CUSTOM)
+ .withContent(content)
+ .withEnabled(true)
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ policyMetaService.insertPolicy(policyEntity, false);
+
+ assertEquals(policyEntity,
policyMetaService.getPolicyById(policyEntity.id()));
+ }
+
@TestTemplate
public void testInsertAndGetPolicyByIdentifier() throws IOException {
BaseMetalake metalake = createAndInsertMakeLake(METALAKE_NAME);
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestTableMetaService.java
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestTableMetaService.java
index 7a6b87e797..4e4fc85f34 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestTableMetaService.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestTableMetaService.java
@@ -99,6 +99,24 @@ public class TestTableMetaService extends TestJDBCBackend {
EntityChangeLogMapper.class, mapper ->
mapper.selectEntityChanges(lastConsumedId, 100));
}
+ @TestTemplate
+ public void testGetMetalakeNameByTableId() throws IOException {
+ createAndInsertMakeLake(metalakeName);
+ createAndInsertCatalog(metalakeName, catalogName);
+ createAndInsertSchema(metalakeName, catalogName, schemaName);
+ TableEntity table =
+ createTableEntity(
+ RandomIdGenerator.INSTANCE.nextId(),
+ NamespaceUtil.ofTable(metalakeName, catalogName, schemaName),
+ "table_metalake_by_id",
+ AUDIT_INFO);
+ TableMetaService.getInstance().insertTable(table, false);
+
+ Assertions.assertEquals(
+ metalakeName,
TableMetaService.getInstance().getMetalakeNameByTableId(table.id()));
+
Assertions.assertNull(TableMetaService.getInstance().getMetalakeNameByTableId(table.id()
+ 1));
+ }
+
@TestTemplate
public void testLegacyTimelineDeleteKeepsLiveVersionsOfSameTable() throws
Exception {
createAndInsertMakeLake(metalakeName);
diff --git a/maintenance/optimizer-api/build.gradle.kts
b/maintenance/optimizer-api/build.gradle.kts
index 07d507dfcd..4af4819baa 100644
--- a/maintenance/optimizer-api/build.gradle.kts
+++ b/maintenance/optimizer-api/build.gradle.kts
@@ -31,15 +31,18 @@ dependencies {
implementation(libs.guava)
implementation(libs.commons.lang3)
implementation(libs.jackson.databind)
+ implementation(libs.ql.expression)
annotationProcessor(libs.lombok)
compileOnly(libs.lombok)
+ compileOnly(libs.slf4j.api)
testAnnotationProcessor(libs.lombok)
testCompileOnly(libs.lombok)
testImplementation(libs.junit.jupiter.api)
testImplementation(libs.mockito.core)
testRuntimeOnly(libs.junit.jupiter.engine)
+ testRuntimeOnly("org.slf4j:slf4j-simple:1.7.36")
}
tasks.test {
diff --git
a/maintenance/optimizer-api/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/compaction/CompactionTriggerEvaluator.java
b/maintenance/optimizer-api/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/compaction/CompactionTriggerEvaluator.java
new file mode 100644
index 0000000000..e77a1517ba
--- /dev/null
+++
b/maintenance/optimizer-api/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/compaction/CompactionTriggerEvaluator.java
@@ -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.
+ */
+package
org.apache.gravitino.maintenance.optimizer.recommender.handler.compaction;
+
+import com.google.common.base.Preconditions;
+import java.util.HashMap;
+import java.util.Map;
+import
org.apache.gravitino.maintenance.optimizer.recommender.util.ExpressionEvaluator;
+import
org.apache.gravitino.maintenance.optimizer.recommender.util.QLExpressionEvaluator;
+import org.apache.gravitino.policy.IcebergDataCompactionContent;
+
+/**
+ * Evaluates Iceberg compaction {@code trigger-expr} with the same QL
expression stack used by the
+ * optimizer compaction strategy handler ({@code CompactionStrategyHandler} /
{@code
+ * BaseExpressionStrategyHandler}).
+ *
+ * <p>Intended for callers that already have table-level metric values (for
example TMS in-job
+ * sample JSON) without constructing a full strategy handler.
+ */
+public final class CompactionTriggerEvaluator {
+
+ private static final ExpressionEvaluator EVALUATOR = new
QLExpressionEvaluator();
+
+ private CompactionTriggerEvaluator() {}
+
+ /**
+ * @param content compaction policy content (provides numeric rules +
trigger-expr)
+ * @param metricValues sampled metric name → value (e.g. {@code
custom-data-file-mse})
+ * @return whether maintenance should run
+ */
+ public static boolean shouldMaintain(
+ IcebergDataCompactionContent content, Map<String, Object> metricValues) {
+ Preconditions.checkNotNull(content, "content");
+ Preconditions.checkNotNull(metricValues, "metricValues");
+
+ Map<String, Object> context = new HashMap<>();
+ // Mirror BaseExpressionStrategyHandler#getRulesExpressionContext: numeric
rules only.
+ content
+ .rules()
+ .forEach(
+ (key, value) -> {
+ if (value == null) {
+ return;
+ }
+ try {
+ context.put(key, Long.parseLong(value.toString()));
+ } catch (NumberFormatException ignored) {
+ // Non-numeric rule values (expressions, strings) are not QL
bindings.
+ }
+ });
+ context.putAll(metricValues);
+
+ Object trigger =
content.rules().get(IcebergDataCompactionContent.TRIGGER_EXPR_KEY);
+ String expression =
+ trigger == null || trigger.toString().trim().isEmpty()
+ ? "false"
+ : trigger.toString().trim();
+ return EVALUATOR.evaluateBool(expression, context);
+ }
+}
diff --git
a/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/util/ExpressionEvaluator.java
b/maintenance/optimizer-api/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/util/ExpressionEvaluator.java
similarity index 100%
rename from
maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/util/ExpressionEvaluator.java
rename to
maintenance/optimizer-api/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/util/ExpressionEvaluator.java
diff --git
a/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/util/QLExpressionEvaluator.java
b/maintenance/optimizer-api/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/util/QLExpressionEvaluator.java
similarity index 98%
rename from
maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/util/QLExpressionEvaluator.java
rename to
maintenance/optimizer-api/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/util/QLExpressionEvaluator.java
index 1a547aeac6..d74e874020 100644
---
a/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/util/QLExpressionEvaluator.java
+++
b/maintenance/optimizer-api/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/util/QLExpressionEvaluator.java
@@ -36,6 +36,10 @@ import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+/**
+ * QLExpress-backed {@link ExpressionEvaluator} used by optimizer strategy
handlers and TMS in-job
+ * recommenders.
+ */
public class QLExpressionEvaluator implements ExpressionEvaluator {
private static final Logger LOG =
LoggerFactory.getLogger(QLExpressionEvaluator.class);
diff --git a/plugins/tms/build.gradle.kts b/plugins/tms/build.gradle.kts
index b12be69734..33b8a15efc 100644
--- a/plugins/tms/build.gradle.kts
+++ b/plugins/tms/build.gradle.kts
@@ -29,11 +29,14 @@ dependencies {
implementation(project(":api"))
implementation(project(":common"))
implementation(project(":core"))
+ implementation(project(":maintenance:optimizer-api"))
implementation(project(":server-common"))
implementation(libs.bundles.jersey)
+ implementation(libs.commons.lang3)
implementation(libs.db.scheduler)
implementation(libs.guava)
+ implementation(libs.metrics.jersey2)
implementation(libs.mybatis)
implementation(libs.servlet)
diff --git
a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/injob/InJobRecommender.java
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/injob/AlwaysMaintainInJobRecommender.java
similarity index 55%
copy from
plugins/tms/src/main/java/org/apache/gravitino/maintenance/injob/InJobRecommender.java
copy to
plugins/tms/src/main/java/org/apache/gravitino/maintenance/injob/AlwaysMaintainInJobRecommender.java
index dd05f934e7..188bd7643c 100644
---
a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/injob/InJobRecommender.java
+++
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/injob/AlwaysMaintainInJobRecommender.java
@@ -19,21 +19,23 @@
package org.apache.gravitino.maintenance.injob;
/**
- * Decides whether maintenance should run after metrics were sampled at {@code
S0}.
+ * In-job recommender that always returns {@link RecommendDecision#MAINTAIN}.
*
- * <p>Production implementations come with the four maintenance executors;
callers inject the
- * strategy for the current type.
+ * <p>Production {@link TypeDispatchingInJobRecommender} uses this for expire
/ rewrite-manifests
+ * placeholders: cron already scheduled the work, while policy content still
ships {@code
+ * trigger-expr=false} for optimizer-only gating. Also available for tests and
injectable overrides.
*/
-@FunctionalInterface
-public interface InJobRecommender {
+public final class AlwaysMaintainInJobRecommender implements InJobRecommender {
- /**
- * @param tableId table entity id
- * @param policyId policy entity id
- * @param sampledSnapshotId Iceberg snapshot id at sample time
- * @param metricsJson serialized metrics for that snapshot
- * @return maintain or skip
- */
- RecommendDecision recommend(
- long tableId, long policyId, long sampledSnapshotId, String metricsJson);
+ /** Shared instance. */
+ public static final AlwaysMaintainInJobRecommender INSTANCE =
+ new AlwaysMaintainInJobRecommender();
+
+ private AlwaysMaintainInJobRecommender() {}
+
+ @Override
+ public RecommendDecision recommend(
+ long tableId, long policyId, long sampledSnapshotId, String metricsJson)
{
+ return RecommendDecision.MAINTAIN;
+ }
}
diff --git
a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/injob/CompactionInJobRecommender.java
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/injob/CompactionInJobRecommender.java
new file mode 100644
index 0000000000..ebf33c72f9
--- /dev/null
+++
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/injob/CompactionInJobRecommender.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.maintenance.injob;
+
+import com.google.common.base.Preconditions;
+import java.util.Map;
+import java.util.Optional;
+import java.util.function.Function;
+import javax.annotation.Nullable;
+import
org.apache.gravitino.maintenance.optimizer.recommender.handler.compaction.CompactionTriggerEvaluator;
+import org.apache.gravitino.meta.PolicyEntity;
+import org.apache.gravitino.policy.IcebergDataCompactionContent;
+import org.apache.gravitino.policy.PolicyContent;
+import org.apache.gravitino.storage.relational.service.PolicyMetaService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * In-job recommender for Iceberg compaction. Evaluates {@code trigger-expr}
via {@link
+ * CompactionTriggerEvaluator} (same QL path as the optimizer strategy
handler).
+ */
+public final class CompactionInJobRecommender implements InJobRecommender {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(CompactionInJobRecommender.class);
+ private static final PolicyMetaService POLICY_SERVICE =
PolicyMetaService.getInstance();
+
+ private final Function<Long, Optional<PolicyEntity>> policyLoader;
+
+ /** Loads policies through {@link PolicyMetaService}. */
+ public CompactionInJobRecommender() {
+ this(CompactionInJobRecommender::loadPolicy);
+ }
+
+ /**
+ * @param policyLoader loads a policy by id when the caller did not supply a
policy entity
+ */
+ CompactionInJobRecommender(Function<Long, Optional<PolicyEntity>>
policyLoader) {
+ this.policyLoader = Preconditions.checkNotNull(policyLoader);
+ }
+
+ @Override
+ public RecommendDecision recommend(
+ long tableId, long policyId, long sampledSnapshotId, String metricsJson)
{
+ return recommend(tableId, policyId, sampledSnapshotId, metricsJson, null);
+ }
+
+ @Override
+ public RecommendDecision recommend(
+ long tableId,
+ long policyId,
+ long sampledSnapshotId,
+ String metricsJson,
+ @Nullable PolicyEntity policy) {
+ IcebergDataCompactionContent compaction =
resolveCompactionContent(tableId, policyId, policy);
+ if (compaction == null) {
+ return RecommendDecision.SKIP;
+ }
+ Map<String, Object> metrics = MetricsJson.parseNumericValues(metricsJson);
+ if (metrics.isEmpty()
+ ||
!metrics.containsKey(IcebergDataCompactionContent.DATA_FILE_MSE_METRIC)
+ ||
!metrics.containsKey(IcebergDataCompactionContent.DELETE_FILE_NUMBER_METRIC)) {
+ LOG.warn(
+ "Compaction recommend skip: metrics JSON missing compaction fields
for table {} policy {}",
+ tableId,
+ policyId);
+ return RecommendDecision.SKIP;
+ }
+ boolean maintain = CompactionTriggerEvaluator.shouldMaintain(compaction,
metrics);
+ LOG.info(
+ "Compaction recommend table={} policy={} snapshot={} metrics={} -> {}",
+ tableId,
+ policyId,
+ sampledSnapshotId,
+ metrics,
+ maintain ? "MAINTAIN" : "SKIP");
+ return maintain ? RecommendDecision.MAINTAIN : RecommendDecision.SKIP;
+ }
+
+ @Nullable
+ private IcebergDataCompactionContent resolveCompactionContent(
+ long tableId, long policyId, @Nullable PolicyEntity policy) {
+ PolicyContent content;
+ if (policy != null) {
+ content = policy.content();
+ } else {
+ Optional<PolicyEntity> loaded = policyLoader.apply(policyId);
+ if (loaded.isEmpty()) {
+ LOG.warn("Compaction recommend skip: policy {} not found (table={})",
policyId, tableId);
+ return null;
+ }
+ content = loaded.get().content();
+ }
+ if (!(content instanceof IcebergDataCompactionContent)) {
+ LOG.warn(
+ "Compaction recommend skip: policy {} is not
IcebergDataCompactionContent (table={})",
+ policyId,
+ tableId);
+ return null;
+ }
+ return (IcebergDataCompactionContent) content;
+ }
+
+ private static Optional<PolicyEntity> loadPolicy(long policyId) {
+ return Optional.ofNullable(POLICY_SERVICE.getPolicyById(policyId));
+ }
+}
diff --git
a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/injob/InJobRecommender.java
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/injob/InJobRecommender.java
index dd05f934e7..7425e020ca 100644
---
a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/injob/InJobRecommender.java
+++
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/injob/InJobRecommender.java
@@ -18,13 +18,16 @@
*/
package org.apache.gravitino.maintenance.injob;
+import javax.annotation.Nullable;
+import org.apache.gravitino.meta.PolicyEntity;
+
/**
* Decides whether maintenance should run after metrics were sampled at {@code
S0}.
*
- * <p>Production implementations come with the four maintenance executors;
callers inject the
- * strategy for the current type.
+ * <p>Production wiring uses {@link TypeDispatchingInJobRecommender}:
compaction via {@link
+ * CompactionInJobRecommender}, orphan cleanup via {@link
TriggerExprInJobRecommender}, and expire /
+ * rewrite-manifests placeholders via {@link AlwaysMaintainInJobRecommender}.
*/
-@FunctionalInterface
public interface InJobRecommender {
/**
@@ -36,4 +39,24 @@ public interface InJobRecommender {
*/
RecommendDecision recommend(
long tableId, long policyId, long sampledSnapshotId, String metricsJson);
+
+ /**
+ * Same as {@link #recommend(long, long, long, String)} when the caller
already loaded the policy
+ * (avoids a second store round-trip in type-dispatching recommenders).
+ *
+ * @param tableId table entity id
+ * @param policyId policy entity id
+ * @param sampledSnapshotId Iceberg snapshot id at sample time
+ * @param metricsJson serialized metrics for that snapshot
+ * @param policy already-loaded policy, or {@code null} to fall back to id
lookup
+ * @return maintain or skip
+ */
+ default RecommendDecision recommend(
+ long tableId,
+ long policyId,
+ long sampledSnapshotId,
+ String metricsJson,
+ @Nullable PolicyEntity policy) {
+ return recommend(tableId, policyId, sampledSnapshotId, metricsJson);
+ }
}
diff --git
a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/injob/MetricsJson.java
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/injob/MetricsJson.java
new file mode 100644
index 0000000000..9fd61fb487
--- /dev/null
+++
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/injob/MetricsJson.java
@@ -0,0 +1,70 @@
+/*
+ * 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.injob;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Map.Entry;
+import javax.annotation.Nullable;
+import org.apache.gravitino.json.JsonUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** Parses numeric fields from TMS in-job metrics JSON payloads. */
+final class MetricsJson {
+
+ private static final Logger LOG = LoggerFactory.getLogger(MetricsJson.class);
+
+ private MetricsJson() {}
+
+ /**
+ * @param metricsJson JSON object, possibly null/blank
+ * @return numeric field bindings (integral → {@link Long}, otherwise {@link
Double}); empty on
+ * parse failure
+ */
+ static Map<String, Object> parseNumericValues(@Nullable String metricsJson) {
+ Map<String, Object> metrics = new HashMap<>();
+ if (metricsJson == null || metricsJson.trim().isEmpty()) {
+ return metrics;
+ }
+ try {
+ JsonNode root = JsonUtils.anyFieldMapper().readTree(metricsJson);
+ Iterator<Entry<String, JsonNode>> fields = root.fields();
+ while (fields.hasNext()) {
+ Entry<String, JsonNode> field = fields.next();
+ JsonNode value = field.getValue();
+ if (value == null || value.isNull() || !value.isNumber()) {
+ continue;
+ }
+ if (value.isIntegralNumber()) {
+ metrics.put(field.getKey(), value.longValue());
+ } else {
+ metrics.put(field.getKey(), value.doubleValue());
+ }
+ }
+ return metrics;
+ } catch (IOException | RuntimeException e) {
+ LOG.warn("Failed to parse metrics JSON: {}", e.toString());
+ return metrics;
+ }
+ }
+}
diff --git
a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/injob/RecommendDecision.java
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/injob/RecommendDecision.java
index eff0d58650..9c337fd40d 100644
---
a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/injob/RecommendDecision.java
+++
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/injob/RecommendDecision.java
@@ -20,10 +20,7 @@ package org.apache.gravitino.maintenance.injob;
/**
* Whether the in-job path should run maintenance work after a successful
{@code S0} sample
- * (§5.2.2).
- *
- * <p>Type-specific strategy recommenders are wired in a follow-up subtask;
this enum only models
- * the gate outcome.
+ * (§5.2.2). Produced by {@link TypeDispatchingInJobRecommender} and
type-specific recommenders.
*/
public enum RecommendDecision {
/** Proceed to execute maintenance work. */
diff --git
a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/injob/TriggerExprInJobRecommender.java
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/injob/TriggerExprInJobRecommender.java
new file mode 100644
index 0000000000..a5ee5a28e8
--- /dev/null
+++
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/injob/TriggerExprInJobRecommender.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.injob;
+
+import com.google.common.base.Preconditions;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.function.Function;
+import javax.annotation.Nullable;
+import
org.apache.gravitino.maintenance.optimizer.recommender.util.QLExpressionEvaluator;
+import org.apache.gravitino.meta.PolicyEntity;
+import org.apache.gravitino.policy.IcebergOrphanFileRemovalContent;
+import org.apache.gravitino.policy.PolicyContent;
+import org.apache.gravitino.storage.relational.service.PolicyMetaService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * In-job recommender that evaluates policy {@code trigger-expr} with the
shared QL stack (numeric
+ * rules + sampled metric bindings). Used for types whose content already
exposes a real
+ * trigger-expr (for example orphan cleanup {@code true}). Expire /
rewrite-manifests placeholders
+ * are dispatched to {@link AlwaysMaintainInJobRecommender} instead.
+ */
+public final class TriggerExprInJobRecommender implements InJobRecommender {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(TriggerExprInJobRecommender.class);
+ private static final PolicyMetaService POLICY_SERVICE =
PolicyMetaService.getInstance();
+
+ private final Function<Long, Optional<PolicyEntity>> policyLoader;
+ private final QLExpressionEvaluator evaluator;
+
+ /** Loads policies through {@link PolicyMetaService}. */
+ public TriggerExprInJobRecommender() {
+ this(TriggerExprInJobRecommender::loadPolicy);
+ }
+
+ /**
+ * @param policyLoader loads a policy by id when the caller did not supply a
policy entity
+ */
+ TriggerExprInJobRecommender(Function<Long, Optional<PolicyEntity>>
policyLoader) {
+ this.policyLoader = Preconditions.checkNotNull(policyLoader);
+ this.evaluator = new QLExpressionEvaluator();
+ }
+
+ @Override
+ public RecommendDecision recommend(
+ long tableId, long policyId, long sampledSnapshotId, String metricsJson)
{
+ return recommend(tableId, policyId, sampledSnapshotId, metricsJson, null);
+ }
+
+ @Override
+ public RecommendDecision recommend(
+ long tableId,
+ long policyId,
+ long sampledSnapshotId,
+ String metricsJson,
+ @Nullable PolicyEntity policy) {
+ PolicyContent content = resolveContent(tableId, policyId, policy);
+ if (content == null) {
+ return RecommendDecision.SKIP;
+ }
+ Object trigger =
content.rules().get(IcebergOrphanFileRemovalContent.TRIGGER_EXPR_KEY);
+ String expression =
+ trigger == null || trigger.toString().trim().isEmpty()
+ ? "false"
+ : trigger.toString().trim();
+
+ Map<String, Object> context = new HashMap<>();
+ content
+ .rules()
+ .forEach(
+ (key, value) -> {
+ if (value == null) {
+ return;
+ }
+ try {
+ context.put(key, Long.parseLong(value.toString()));
+ } catch (NumberFormatException ignored) {
+ // Non-numeric rule values are not QL bindings.
+ }
+ });
+ context.putAll(MetricsJson.parseNumericValues(metricsJson));
+
+ boolean maintain = evaluator.evaluateBool(expression, context);
+ LOG.info(
+ "Trigger-expr recommend table={} policy={} snapshot={} expr={} -> {}",
+ tableId,
+ policyId,
+ sampledSnapshotId,
+ expression,
+ maintain ? "MAINTAIN" : "SKIP");
+ return maintain ? RecommendDecision.MAINTAIN : RecommendDecision.SKIP;
+ }
+
+ @Nullable
+ private PolicyContent resolveContent(long tableId, long policyId, @Nullable
PolicyEntity policy) {
+ if (policy != null) {
+ return policy.content();
+ }
+ Optional<PolicyEntity> loaded = policyLoader.apply(policyId);
+ if (loaded.isEmpty()) {
+ LOG.warn("Trigger-expr recommend skip: policy {} not found (table={})",
policyId, tableId);
+ return null;
+ }
+ return loaded.get().content();
+ }
+
+ private static Optional<PolicyEntity> loadPolicy(long policyId) {
+ return Optional.ofNullable(POLICY_SERVICE.getPolicyById(policyId));
+ }
+}
diff --git
a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/injob/TypeDispatchingInJobRecommender.java
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/injob/TypeDispatchingInJobRecommender.java
new file mode 100644
index 0000000000..be45a544fb
--- /dev/null
+++
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/injob/TypeDispatchingInJobRecommender.java
@@ -0,0 +1,138 @@
+/*
+ * 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.injob;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import java.util.Optional;
+import java.util.function.Function;
+import org.apache.gravitino.meta.PolicyEntity;
+import org.apache.gravitino.policy.IcebergDataCompactionContent;
+import org.apache.gravitino.policy.IcebergRewriteManifestsContent;
+import org.apache.gravitino.policy.IcebergSnapshotExpirationContent;
+import org.apache.gravitino.policy.Policy;
+import org.apache.gravitino.storage.relational.service.PolicyMetaService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Dispatches in-job recommenders by policy type.
+ *
+ * <ul>
+ * <li>Compaction → threshold / {@code trigger-expr} evaluation
+ * <li>Orphan cleanup → policy {@code trigger-expr} ({@code true} by default)
+ * <li>Expire / rewrite-manifests placeholders → always maintain (cron
already scheduled the work;
+ * content still ships {@code trigger-expr=false} for optimizer-only
gating)
+ * </ul>
+ */
+public final class TypeDispatchingInJobRecommender implements InJobRecommender
{
+
+ private static final Logger LOG =
LoggerFactory.getLogger(TypeDispatchingInJobRecommender.class);
+ private static final PolicyMetaService POLICY_SERVICE =
PolicyMetaService.getInstance();
+
+ private final Function<Long, Optional<PolicyEntity>> policyLoader;
+ private final InJobRecommender compactionRecommender;
+ private final InJobRecommender triggerExprRecommender;
+ private final InJobRecommender placeholderRecommender;
+
+ /** Production wiring. Loads policies through {@link PolicyMetaService}. */
+ public TypeDispatchingInJobRecommender() {
+ this(TypeDispatchingInJobRecommender::loadPolicy);
+ }
+
+ private TypeDispatchingInJobRecommender(Function<Long,
Optional<PolicyEntity>> policyLoader) {
+ this(
+ policyLoader,
+ new CompactionInJobRecommender(policyLoader),
+ new TriggerExprInJobRecommender(policyLoader),
+ AlwaysMaintainInJobRecommender.INSTANCE);
+ }
+
+ /**
+ * @param policyLoader loads a policy by id
+ * @param compactionRecommender compaction thresholds
+ * @param triggerExprRecommender trigger-expr evaluation (orphan and similar)
+ * @param placeholderRecommender expire / rewrite-manifests placeholders
until real thresholds
+ * land
+ */
+ TypeDispatchingInJobRecommender(
+ Function<Long, Optional<PolicyEntity>> policyLoader,
+ InJobRecommender compactionRecommender,
+ InJobRecommender triggerExprRecommender,
+ InJobRecommender placeholderRecommender) {
+ this.policyLoader = Preconditions.checkNotNull(policyLoader);
+ this.compactionRecommender =
Preconditions.checkNotNull(compactionRecommender);
+ this.triggerExprRecommender =
Preconditions.checkNotNull(triggerExprRecommender);
+ this.placeholderRecommender =
Preconditions.checkNotNull(placeholderRecommender);
+ }
+
+ /**
+ * Test wiring that routes expire / rewrite-manifests placeholders to {@link
+ * AlwaysMaintainInJobRecommender#INSTANCE}.
+ *
+ * @param policyLoader loads a policy by id
+ * @param compactionRecommender compaction thresholds
+ * @param triggerExprRecommender trigger-expr evaluation for orphan and
similar
+ */
+ @VisibleForTesting
+ TypeDispatchingInJobRecommender(
+ Function<Long, Optional<PolicyEntity>> policyLoader,
+ InJobRecommender compactionRecommender,
+ InJobRecommender triggerExprRecommender) {
+ this(
+ policyLoader,
+ compactionRecommender,
+ triggerExprRecommender,
+ AlwaysMaintainInJobRecommender.INSTANCE);
+ }
+
+ @Override
+ public RecommendDecision recommend(
+ long tableId, long policyId, long sampledSnapshotId, String metricsJson)
{
+ Optional<PolicyEntity> loaded = policyLoader.apply(policyId);
+ if (loaded.isEmpty()) {
+ LOG.warn("In-job recommend: policy {} missing; skip (table={})",
policyId, tableId);
+ return RecommendDecision.SKIP;
+ }
+ PolicyEntity policy = loaded.get();
+ if (policy.policyType() == Policy.BuiltInType.ICEBERG_COMPACTION
+ || policy.content() instanceof IcebergDataCompactionContent) {
+ return compactionRecommender.recommend(
+ tableId, policyId, sampledSnapshotId, metricsJson, policy);
+ }
+ if (isPlaceholderMaintenance(policy)) {
+ return placeholderRecommender.recommend(
+ tableId, policyId, sampledSnapshotId, metricsJson, policy);
+ }
+ return triggerExprRecommender.recommend(
+ tableId, policyId, sampledSnapshotId, metricsJson, policy);
+ }
+
+ private static boolean isPlaceholderMaintenance(PolicyEntity policy) {
+ Policy.BuiltInType type = policy.policyType();
+ return type == Policy.BuiltInType.ICEBERG_SNAPSHOT_EXPIRATION
+ || type == Policy.BuiltInType.ICEBERG_REWRITE_MANIFESTS
+ || policy.content() instanceof IcebergSnapshotExpirationContent
+ || policy.content() instanceof IcebergRewriteManifestsContent;
+ }
+
+ private static Optional<PolicyEntity> loadPolicy(long policyId) {
+ return Optional.ofNullable(POLICY_SERVICE.getPolicyById(policyId));
+ }
+}
diff --git
a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/occupancy/OccupancyGate.java
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/occupancy/OccupancyGate.java
index 156d75047f..cf31ca7502 100644
---
a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/occupancy/OccupancyGate.java
+++
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/occupancy/OccupancyGate.java
@@ -19,9 +19,7 @@
package org.apache.gravitino.maintenance.occupancy;
import com.google.common.base.Preconditions;
-import java.util.Collections;
import java.util.EnumSet;
-import java.util.List;
import java.util.Set;
import javax.annotation.Nullable;
import org.apache.gravitino.exceptions.NoSuchEntityException;
@@ -30,26 +28,33 @@ import
org.apache.gravitino.maintenance.TableMaintenanceConfig;
import org.apache.gravitino.maintenance.storage.po.TableMaintenanceActionPO;
import
org.apache.gravitino.maintenance.storage.service.TableMaintenanceActionMetaService;
import org.apache.gravitino.meta.JobEntity;
-import org.apache.gravitino.storage.relational.mapper.MetalakeMetaMapper;
-import org.apache.gravitino.storage.relational.mapper.TableMetaMapper;
-import org.apache.gravitino.storage.relational.po.MetalakePO;
-import org.apache.gravitino.storage.relational.po.TablePO;
import org.apache.gravitino.storage.relational.service.JobMetaService;
-import org.apache.gravitino.storage.relational.utils.SessionUtils;
+import org.apache.gravitino.storage.relational.service.TableMetaService;
import org.apache.gravitino.utils.NameIdentifierUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* Expand / submit occupancy checks against {@code table_maintenance_action}
(§5.2).
*
* <p>Unbound rows ({@code job_id} null, {@code finished_at = 0}) are a short
claim: occupied while
* still within TTL on the <strong>database</strong> clock (same timeline as
acquire/reclaim). Bound
- * rows consult {@link JobMetaService} and do <strong>not</strong> backfill
{@code finished_at}.
+ * rows consult {@link JobMetaService}; when that job is already terminal or
confirmed missing,
+ * {@link #reclaimTerminalBound} marks {@code finished_at} so the soft slot
can be claimed again
+ * (without inventing stub S1 metrics rows or copying {@code
before_snapshot_id} into {@code
+ * after_snapshot_id}). A later finish listener may still backfill a real S1
via {@link
+ * TableMaintenanceActionMetaService#markFinished}.
*/
public final class OccupancyGate {
+ private static final Logger LOG =
LoggerFactory.getLogger(OccupancyGate.class);
+
private static final Set<JobHandle.Status> ACTIVE =
EnumSet.of(JobHandle.Status.QUEUED, JobHandle.Status.STARTED,
JobHandle.Status.CANCELLING);
+ private static final JobMetaService JOB_SERVICE =
JobMetaService.getInstance();
+ private static final TableMetaService TABLE_SERVICE =
TableMetaService.getInstance();
+
private final TableMaintenanceActionMetaService actionMetaService;
private final long claimTimeoutMs;
@@ -78,22 +83,74 @@ public final class OccupancyGate {
/**
* @param tableId table entity id
* @param policyId policy entity id
- * @return true when any open row is a live unbound claim or an active
Gravitino job
+ * @return true when any open row is a live unbound claim or a still-bound
soft-slot row
*/
public boolean isInFlight(long tableId, long policyId) {
+ reclaimTerminalBound(tableId, policyId);
if (actionMetaService.hasLiveUnboundClaim(tableId, policyId,
claimTimeoutMs)) {
return true;
}
+ // After reclaim, any remaining bound open row occupies the soft slot
(ACTIVE job, unresolved
+ // lookup, or a terminal/missing job whose markFinished raced/failed).
Fail closed so
+ // UNIQUE(table, policy, finished_at=0) cannot be ignored while still
blocking
+ // tryAcquireInFlight.
+ for (TableMaintenanceActionPO open :
actionMetaService.listInFlight(tableId, policyId)) {
+ if (open.jobId() != null) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Job-terminal reconciler failsafe (§5.2.2): when Spark finish callback was
missed, mark the
+ * action finished so the soft slot can be reused. Does not invent stub
metrics rows (those would
+ * pollute {@code table_snapshot_metrics}), and does not copy {@code
before_snapshot_id} into
+ * {@code after_snapshot_id} — only a real S1 (or null) is stored (§5.2.3).
+ *
+ * <p>Only reclaim when the bound job is confirmed terminal or confirmed
missing. An unresolved
+ * job lookup (for example the table/metalake cannot be resolved) leaves the
row open.
+ *
+ * @param tableId table entity id
+ * @param policyId policy entity id
+ */
+ public void reclaimTerminalBound(long tableId, long policyId) {
for (TableMaintenanceActionPO open :
actionMetaService.listInFlight(tableId, policyId)) {
if (open.jobId() == null) {
continue;
}
- JobEntity job = loadJob(open.jobId(), tableId);
- if (job != null && ACTIVE.contains(job.status())) {
- return true;
+ BoundJobLookup lookup = lookupBoundJob(open.jobId(), tableId);
+ if (lookup.kind == BoundJobLookup.Kind.ACTIVE
+ || lookup.kind == BoundJobLookup.Kind.UNRESOLVED) {
+ if (lookup.kind == BoundJobLookup.Kind.UNRESOLVED) {
+ LOG.warn(
+ "Skipping reclaim of job {} for table {} policy {}: metalake
could not be resolved",
+ open.jobId(),
+ tableId,
+ policyId);
+ }
+ continue;
+ }
+ JobEntity job = lookup.job;
+ long finishedAt =
+ job != null && job.finishedAt() != null && job.finishedAt() > 0
+ ? job.finishedAt()
+ : System.currentTimeMillis();
+ // Keep a real after snapshot only; never invent S1 from S0.
+ Long afterSnapshotId = open.afterSnapshotId();
+ try {
+ actionMetaService.markFinished(open.jobId(), afterSnapshotId,
finishedAt);
+ } catch (RuntimeException e) {
+ // Another node may have finished the row, or DB is unhealthy — log
and keep fail-closed
+ // via isInFlight seeing the still-open bound row.
+ LOG.warn(
+ "Failed to reclaim terminal/missing job {} for table {} policy {}:
{}",
+ open.jobId(),
+ tableId,
+ policyId,
+ e.toString());
}
}
- return false;
}
/**
@@ -136,11 +193,14 @@ public final class OccupancyGate {
if (open.jobId() == null) {
continue;
}
- JobEntity job = loadJob(open.jobId(), tableId);
- if (job == null || ACTIVE.contains(job.status())) {
+ BoundJobLookup lookup = lookupBoundJob(open.jobId(), tableId);
+ if (lookup.kind != BoundJobLookup.Kind.TERMINAL || lookup.job == null) {
continue;
}
- Long fromJob = job.finishedAt() != null && job.finishedAt() > 0 ?
job.finishedAt() : null;
+ Long fromJob =
+ lookup.job.finishedAt() != null && lookup.job.finishedAt() > 0
+ ? lookup.job.finishedAt()
+ : null;
if (fromJob == null) {
continue;
}
@@ -151,33 +211,63 @@ public final class OccupancyGate {
return latest;
}
- @Nullable
- private static JobEntity loadJob(long jobId, long tableId) {
- String metalake = metalakeNameForTable(tableId);
+ /**
+ * Resolves the bound job through the table's metalake, not the current
policy. A deleted policy
+ * must not make a still-running job look "missing".
+ */
+ private static BoundJobLookup lookupBoundJob(long jobId, long tableId) {
+ String metalake = TABLE_SERVICE.getMetalakeNameByTableId(tableId);
if (metalake == null) {
- return null;
+ return BoundJobLookup.unresolved();
}
try {
- return JobMetaService.getInstance()
- .getJobByIdentifier(NameIdentifierUtil.ofJob(metalake,
JobHandle.JOB_ID_PREFIX + jobId));
+ JobEntity job =
+ JOB_SERVICE.getJobByIdentifier(
+ NameIdentifierUtil.ofJob(metalake, JobHandle.JOB_ID_PREFIX +
jobId));
+ if (ACTIVE.contains(job.status())) {
+ return BoundJobLookup.active(job);
+ }
+ return BoundJobLookup.terminal(job);
} catch (NoSuchEntityException e) {
- return null;
+ return BoundJobLookup.missing();
}
}
- @Nullable
- private static String metalakeNameForTable(long tableId) {
- List<TablePO> tables =
- SessionUtils.getWithoutCommit(
- TableMetaMapper.class,
- mapper ->
mapper.listTablePOsByTableIds(Collections.singletonList(tableId)));
- if (tables == null || tables.isEmpty()) {
- return null;
+ /** Outcome of resolving a bound Gravitino job for reclaim / cooldown. */
+ private static final class BoundJobLookup {
+ enum Kind {
+ /** Job meta loaded and still queued/started/cancelling. */
+ ACTIVE,
+ /** Job meta loaded and already terminal. */
+ TERMINAL,
+ /** Metalake resolved but job meta is gone. */
+ MISSING,
+ /** Table/metalake could not be resolved; do not treat as terminal. */
+ UNRESOLVED
+ }
+
+ final Kind kind;
+ @Nullable final JobEntity job;
+
+ private BoundJobLookup(Kind kind, @Nullable JobEntity job) {
+ this.kind = kind;
+ this.job = job;
+ }
+
+ static BoundJobLookup active(JobEntity job) {
+ return new BoundJobLookup(Kind.ACTIVE, job);
+ }
+
+ static BoundJobLookup terminal(JobEntity job) {
+ return new BoundJobLookup(Kind.TERMINAL, job);
+ }
+
+ static BoundJobLookup missing() {
+ return new BoundJobLookup(Kind.MISSING, null);
+ }
+
+ static BoundJobLookup unresolved() {
+ return new BoundJobLookup(Kind.UNRESOLVED, null);
}
- MetalakePO metalake =
- SessionUtils.getWithoutCommit(
- MetalakeMetaMapper.class,
- mapper ->
mapper.selectMetalakeMetaById(tables.get(0).getMetalakeId()));
- return metalake == null ? null : metalake.getMetalakeName();
}
}
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
index 2882964445..f0acccc30a 100644
---
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
@@ -87,6 +87,12 @@ public interface TableMaintenanceActionMapper {
@Param("afterSnapshotId") Long afterSnapshotId,
@Param("finishedAt") long finishedAt);
+ @UpdateProvider(
+ type = TableMaintenanceActionSQLProviderFactory.class,
+ method = "backfillAfterSnapshot")
+ int backfillAfterSnapshot(
+ @Param("jobId") long jobId, @Param("afterSnapshotId") long
afterSnapshotId);
+
@DeleteProvider(type = TableMaintenanceActionSQLProviderFactory.class,
method = "deleteById")
int deleteById(@Param("id") long id);
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
index 891dd120d7..656d1e2585 100644
---
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
@@ -144,6 +144,16 @@ public class TableMaintenanceActionSQLProviderFactory {
return getProvider().markFinished(jobId, afterSnapshotId, finishedAt);
}
+ /**
+ * @param jobId job run id
+ * @param afterSnapshotId after snapshot
+ * @return UPDATE statement
+ */
+ public static String backfillAfterSnapshot(
+ @Param("jobId") long jobId, @Param("afterSnapshotId") long
afterSnapshotId) {
+ return getProvider().backfillAfterSnapshot(jobId, afterSnapshotId);
+ }
+
/**
* @param id action id
* @return DELETE statement
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
index 4ced139557..457ebddb20 100644
---
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
@@ -182,6 +182,22 @@ public class TableMaintenanceActionBaseSQLProvider {
+ " WHERE job_id = #{jobId} AND finished_at = 0";
}
+ /**
+ * Backfills a real S1 after failsafe reclaim closed the soft slot with
{@code after_snapshot_id}
+ * still null (finish listener raced after reclaim).
+ *
+ * @param jobId Gravitino job run id
+ * @param afterSnapshotId terminal snapshot id; must be non-null
+ * @return UPDATE statement
+ */
+ public String backfillAfterSnapshot(
+ @Param("jobId") long jobId, @Param("afterSnapshotId") long
afterSnapshotId) {
+ return "UPDATE "
+ + TABLE_NAME
+ + " SET after_snapshot_id = #{afterSnapshotId}"
+ + " WHERE job_id = #{jobId} AND finished_at > 0 AND after_snapshot_id
IS NULL";
+ }
+
/**
* @param id action row id
* @return DELETE statement
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
index ae1cd03e11..28d47794ec 100644
---
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
@@ -254,6 +254,10 @@ public class TableMaintenanceActionMetaService {
* single soft slot per pair (like {@code deleted_at}). Only one in-flight
row can finish that
* pair at a time, so two completed rows cannot race to the same millisecond
{@code finished_at}.
*
+ * <p>If failsafe reclaim already closed the row ({@code finished_at > 0})
with a null after
+ * snapshot, a later finish callback may still backfill a real S1 when
{@code afterSnapshotId} is
+ * non-null.
+ *
* @param jobId Gravitino job run id
* @param afterSnapshotId terminal snapshot id
* @param finishedAt finish timestamp millis
@@ -267,8 +271,27 @@ public class TableMaintenanceActionMetaService {
SessionUtils.doWithCommitAndFetchResult(
TableMaintenanceActionMapper.class,
mapper -> mapper.markFinished(jobId, afterSnapshotId, finishedAt));
+ if (updated == 1) {
+ return;
+ }
+ if (afterSnapshotId != null) {
+ int backfilled =
+ SessionUtils.doWithCommitAndFetchResult(
+ TableMaintenanceActionMapper.class,
+ mapper -> mapper.backfillAfterSnapshot(jobId, afterSnapshotId));
+ Preconditions.checkState(
+ backfilled == 1,
+ "expected to finish or backfill exactly one action for job_id=%s,
finished=%s, backfilled=%s",
+ jobId,
+ updated,
+ backfilled);
+ return;
+ }
Preconditions.checkState(
- updated == 1, "expected to finish exactly one in-flight action,
updated=%s", updated);
+ false,
+ "expected to finish exactly one in-flight action for job_id=%s,
updated=%s",
+ jobId,
+ updated);
}
/**
diff --git
a/plugins/tms/src/test/java/org/apache/gravitino/maintenance/injob/TestCompactionInJobRecommender.java
b/plugins/tms/src/test/java/org/apache/gravitino/maintenance/injob/TestCompactionInJobRecommender.java
new file mode 100644
index 0000000000..e4b3662199
--- /dev/null
+++
b/plugins/tms/src/test/java/org/apache/gravitino/maintenance/injob/TestCompactionInJobRecommender.java
@@ -0,0 +1,64 @@
+/*
+ * 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.injob;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.Collections;
+import java.util.Optional;
+import org.apache.gravitino.meta.PolicyEntity;
+import org.apache.gravitino.policy.PolicyContents;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+/** Unit tests for {@link CompactionInJobRecommender} (delegates to
CompactionTriggerEvaluator). */
+class TestCompactionInJobRecommender {
+
+ @Test
+ void testRecommendUsesTriggerExpr() {
+ PolicyEntity policy = mock(PolicyEntity.class);
+ when(policy.content())
+ .thenReturn(PolicyContents.icebergDataCompaction(500L, 3L,
Collections.emptyMap()));
+
+ CompactionInJobRecommender recommender =
+ new CompactionInJobRecommender(policyId -> Optional.of(policy));
+ String metrics =
+
"{\"custom-data-file-mse\":600,\"custom-delete-file-number\":0,\"snapshotId\":1}";
+ Assertions.assertEquals(
+ RecommendDecision.MAINTAIN, recommender.recommend(1L, 7L, 99L,
metrics));
+
+ String below =
+
"{\"custom-data-file-mse\":100,\"custom-delete-file-number\":1,\"snapshotId\":1}";
+ Assertions.assertEquals(RecommendDecision.SKIP, recommender.recommend(1L,
7L, 99L, below));
+ }
+
+ @Test
+ void testMissingMetricsSkips() {
+ PolicyEntity policy = mock(PolicyEntity.class);
+ when(policy.content()).thenReturn(PolicyContents.icebergDataCompaction());
+
+ CompactionInJobRecommender recommender =
+ new CompactionInJobRecommender(policyId -> Optional.of(policy));
+ Assertions.assertEquals(
+ RecommendDecision.SKIP,
+ recommender.recommend(
+ 1L, 7L, 99L,
"{\"snapshotId\":1,\"maintenanceType\":\"rewrite_data_files\"}"));
+ }
+}
diff --git
a/plugins/tms/src/test/java/org/apache/gravitino/maintenance/injob/TestMetricsJson.java
b/plugins/tms/src/test/java/org/apache/gravitino/maintenance/injob/TestMetricsJson.java
new file mode 100644
index 0000000000..7b3a328ff2
--- /dev/null
+++
b/plugins/tms/src/test/java/org/apache/gravitino/maintenance/injob/TestMetricsJson.java
@@ -0,0 +1,44 @@
+/*
+ * 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.injob;
+
+import java.util.Map;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+/** Unit tests for {@link MetricsJson}. */
+class TestMetricsJson {
+
+ @Test
+ void testParseNumericValues() {
+ Map<String, Object> metrics =
+ MetricsJson.parseNumericValues(
+
"{\"custom-data-file-mse\":1.5,\"custom-delete-file-number\":3,\"name\":\"x\"}");
+ Assertions.assertEquals(1.5D, metrics.get("custom-data-file-mse"));
+ Assertions.assertEquals(3L, metrics.get("custom-delete-file-number"));
+ Assertions.assertFalse(metrics.containsKey("name"));
+ }
+
+ @Test
+ void testBlankOrInvalidReturnsEmpty() {
+ Assertions.assertTrue(MetricsJson.parseNumericValues(null).isEmpty());
+ Assertions.assertTrue(MetricsJson.parseNumericValues("").isEmpty());
+
Assertions.assertTrue(MetricsJson.parseNumericValues("not-json").isEmpty());
+ }
+}
diff --git
a/plugins/tms/src/test/java/org/apache/gravitino/maintenance/injob/TestTriggerExprInJobRecommender.java
b/plugins/tms/src/test/java/org/apache/gravitino/maintenance/injob/TestTriggerExprInJobRecommender.java
new file mode 100644
index 0000000000..bbf211564e
--- /dev/null
+++
b/plugins/tms/src/test/java/org/apache/gravitino/maintenance/injob/TestTriggerExprInJobRecommender.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.injob;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.Optional;
+import org.apache.gravitino.meta.PolicyEntity;
+import org.apache.gravitino.policy.PolicyContents;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+/** Unit tests for {@link TriggerExprInJobRecommender}. */
+class TestTriggerExprInJobRecommender {
+
+ @Test
+ void testOrphanTriggerTrueMaintains() {
+ PolicyEntity policy = mock(PolicyEntity.class);
+
when(policy.content()).thenReturn(PolicyContents.icebergOrphanFileRemoval());
+
+ TriggerExprInJobRecommender recommender =
+ new TriggerExprInJobRecommender(policyId -> Optional.of(policy));
+ Assertions.assertEquals(
+ RecommendDecision.MAINTAIN, recommender.recommend(2L, 1L, 9L, "{}",
policy));
+ }
+
+ @Test
+ void testExpirePlaceholderTriggerFalseSkips() {
+ PolicyEntity policy = mock(PolicyEntity.class);
+
when(policy.content()).thenReturn(PolicyContents.icebergSnapshotExpiration());
+
+ TriggerExprInJobRecommender recommender =
+ new TriggerExprInJobRecommender(policyId -> Optional.of(policy));
+ Assertions.assertEquals(
+ RecommendDecision.SKIP, recommender.recommend(2L, 1L, 9L, "{}",
policy));
+ }
+
+ @Test
+ void testManifestPlaceholderTriggerFalseSkips() {
+ PolicyEntity policy = mock(PolicyEntity.class);
+
when(policy.content()).thenReturn(PolicyContents.icebergRewriteManifests());
+
+ TriggerExprInJobRecommender recommender =
+ new TriggerExprInJobRecommender(policyId -> Optional.empty());
+ Assertions.assertEquals(
+ RecommendDecision.SKIP, recommender.recommend(2L, 1L, 9L, "{}",
policy));
+ }
+}
diff --git
a/plugins/tms/src/test/java/org/apache/gravitino/maintenance/injob/TestTypeDispatchingInJobRecommender.java
b/plugins/tms/src/test/java/org/apache/gravitino/maintenance/injob/TestTypeDispatchingInJobRecommender.java
new file mode 100644
index 0000000000..836cea899c
--- /dev/null
+++
b/plugins/tms/src/test/java/org/apache/gravitino/maintenance/injob/TestTypeDispatchingInJobRecommender.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.maintenance.injob;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.Collections;
+import java.util.Optional;
+import java.util.function.Function;
+import org.apache.gravitino.meta.PolicyEntity;
+import org.apache.gravitino.policy.Policy;
+import org.apache.gravitino.policy.PolicyContents;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+/** Unit tests for {@link TypeDispatchingInJobRecommender}. */
+class TestTypeDispatchingInJobRecommender {
+
+ @Test
+ void testDispatchesCompaction() {
+ PolicyEntity policy = mock(PolicyEntity.class);
+
when(policy.policyType()).thenReturn(Policy.BuiltInType.ICEBERG_COMPACTION);
+ when(policy.content()).thenReturn(PolicyContents.icebergDataCompaction());
+ Function<Long, Optional<PolicyEntity>> policies =
policiesReturning(policy);
+ InJobRecommender compaction = mock(InJobRecommender.class);
+ when(compaction.recommend(eq(2L), eq(1L), eq(9L), anyString(), eq(policy)))
+ .thenReturn(RecommendDecision.SKIP);
+ InJobRecommender triggerExpr = mock(InJobRecommender.class);
+ InJobRecommender placeholder = mock(InJobRecommender.class);
+
+ TypeDispatchingInJobRecommender dispatcher =
+ new TypeDispatchingInJobRecommender(policies, compaction, triggerExpr,
placeholder);
+ Assertions.assertEquals(RecommendDecision.SKIP, dispatcher.recommend(2L,
1L, 9L, "{\"x\":1}"));
+ verify(compaction).recommend(2L, 1L, 9L, "{\"x\":1}", policy);
+ verify(triggerExpr, never())
+ .recommend(anyLong(), anyLong(), anyLong(), anyString(),
any(PolicyEntity.class));
+ verify(placeholder, never())
+ .recommend(anyLong(), anyLong(), anyLong(), anyString(),
any(PolicyEntity.class));
+ }
+
+ @Test
+ void testDispatchesAlwaysMaintainForExpirePlaceholder() {
+ PolicyEntity policy = mock(PolicyEntity.class);
+
when(policy.policyType()).thenReturn(Policy.BuiltInType.ICEBERG_SNAPSHOT_EXPIRATION);
+
when(policy.content()).thenReturn(PolicyContents.icebergSnapshotExpiration());
+ Function<Long, Optional<PolicyEntity>> policies =
policiesReturning(policy);
+ InJobRecommender compaction = mock(InJobRecommender.class);
+ InJobRecommender triggerExpr = mock(InJobRecommender.class);
+ InJobRecommender placeholder = mock(InJobRecommender.class);
+ when(placeholder.recommend(eq(2L), eq(1L), eq(9L), anyString(),
eq(policy)))
+ .thenReturn(RecommendDecision.MAINTAIN);
+
+ TypeDispatchingInJobRecommender dispatcher =
+ new TypeDispatchingInJobRecommender(policies, compaction, triggerExpr,
placeholder);
+ Assertions.assertEquals(RecommendDecision.MAINTAIN,
dispatcher.recommend(2L, 1L, 9L, "{}"));
+ verify(placeholder).recommend(2L, 1L, 9L, "{}", policy);
+ verify(triggerExpr, never())
+ .recommend(anyLong(), anyLong(), anyLong(), anyString(),
any(PolicyEntity.class));
+ verify(compaction, never())
+ .recommend(anyLong(), anyLong(), anyLong(), anyString(),
any(PolicyEntity.class));
+ }
+
+ @Test
+ void testProductionWiringExpirePlaceholderMaintainsDespiteTriggerFalse() {
+ PolicyEntity policy = mock(PolicyEntity.class);
+
when(policy.policyType()).thenReturn(Policy.BuiltInType.ICEBERG_SNAPSHOT_EXPIRATION);
+
when(policy.content()).thenReturn(PolicyContents.icebergSnapshotExpiration());
+ Function<Long, Optional<PolicyEntity>> policies =
policiesReturning(policy);
+
+ TypeDispatchingInJobRecommender dispatcher =
+ new TypeDispatchingInJobRecommender(
+ policies,
+ new CompactionInJobRecommender(policies),
+ new TriggerExprInJobRecommender(policies),
+ AlwaysMaintainInJobRecommender.INSTANCE);
+
+ Assertions.assertEquals(RecommendDecision.MAINTAIN,
dispatcher.recommend(2L, 1L, 9L, "{}"));
+ }
+
+ @Test
+ void testDispatchesTriggerExprForOrphan() {
+ PolicyEntity policy = mock(PolicyEntity.class);
+
when(policy.policyType()).thenReturn(Policy.BuiltInType.ICEBERG_ORPHAN_FILE_REMOVAL);
+
when(policy.content()).thenReturn(PolicyContents.icebergOrphanFileRemoval());
+ Function<Long, Optional<PolicyEntity>> policies =
policiesReturning(policy);
+
+ TriggerExprInJobRecommender triggerExpr = new
TriggerExprInJobRecommender(policies);
+ TypeDispatchingInJobRecommender dispatcher =
+ new TypeDispatchingInJobRecommender(policies,
mock(InJobRecommender.class), triggerExpr);
+
+ Assertions.assertEquals(RecommendDecision.MAINTAIN,
dispatcher.recommend(2L, 1L, 9L, "{}"));
+ }
+
+ @Test
+ void testCompactionRecommenderReusesLoadedPolicy() {
+ PolicyEntity policy = mock(PolicyEntity.class);
+
when(policy.policyType()).thenReturn(Policy.BuiltInType.ICEBERG_COMPACTION);
+ when(policy.content())
+ .thenReturn(PolicyContents.icebergDataCompaction(500L, 3L,
Collections.emptyMap()));
+ @SuppressWarnings("unchecked")
+ Function<Long, Optional<PolicyEntity>> policies = mock(Function.class);
+ when(policies.apply(7L)).thenReturn(Optional.of(policy));
+
+ CompactionInJobRecommender compaction = new
CompactionInJobRecommender(policies);
+ TypeDispatchingInJobRecommender dispatcher =
+ new TypeDispatchingInJobRecommender(
+ policies, compaction, new TriggerExprInJobRecommender(policies));
+
+ String metrics =
+
"{\"custom-data-file-mse\":600,\"custom-delete-file-number\":0,\"snapshotId\":1}";
+ Assertions.assertEquals(RecommendDecision.MAINTAIN,
dispatcher.recommend(1L, 7L, 99L, metrics));
+ // Type-dispatch loads once; compaction must not look up again.
+ verify(policies, times(1)).apply(7L);
+ }
+
+ @SuppressWarnings("unchecked")
+ private static Function<Long, Optional<PolicyEntity>>
policiesReturning(PolicyEntity policy) {
+ Function<Long, Optional<PolicyEntity>> policies = mock(Function.class);
+ when(policies.apply(anyLong())).thenReturn(Optional.of(policy));
+ return policies;
+ }
+}
diff --git
a/plugins/tms/src/test/java/org/apache/gravitino/maintenance/occupancy/TestOccupancyGate.java
b/plugins/tms/src/test/java/org/apache/gravitino/maintenance/occupancy/TestOccupancyGate.java
index f63532af35..6d0e9cf824 100644
---
a/plugins/tms/src/test/java/org/apache/gravitino/maintenance/occupancy/TestOccupancyGate.java
+++
b/plugins/tms/src/test/java/org/apache/gravitino/maintenance/occupancy/TestOccupancyGate.java
@@ -18,8 +18,11 @@
*/
package org.apache.gravitino.maintenance.occupancy;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
import java.io.IOException;
import java.util.function.Function;
+import org.apache.gravitino.MetadataObject;
import org.apache.gravitino.Namespace;
import org.apache.gravitino.job.JobHandle;
import org.apache.gravitino.maintenance.TableMaintenanceConfig;
@@ -27,7 +30,9 @@ import
org.apache.gravitino.maintenance.storage.service.OccupancyClaim;
import
org.apache.gravitino.maintenance.storage.service.TableMaintenanceActionMetaService;
import org.apache.gravitino.meta.JobEntity;
import org.apache.gravitino.meta.JobTemplateEntity;
+import org.apache.gravitino.meta.PolicyEntity;
import org.apache.gravitino.meta.TableEntity;
+import org.apache.gravitino.policy.PolicyContents;
import org.apache.gravitino.storage.relational.TestJDBCBackend;
import org.apache.gravitino.storage.relational.service.JobMetaService;
import org.apache.gravitino.utils.NameIdentifierUtil;
@@ -38,7 +43,7 @@ import org.junit.jupiter.api.TestTemplate;
class TestOccupancyGate extends TestJDBCBackend {
@TestTemplate
- void testInFlightReadsJobMetaServiceWithoutBackfillingFinishedAt() throws
IOException {
+ void testInFlightReclaimsTerminalBoundJobFinishedAt() throws IOException {
String metalakeName = "tms_occ_ml";
createAndInsertMakeLake(metalakeName);
createAndInsertCatalog(metalakeName, "cat");
@@ -49,7 +54,7 @@ class TestOccupancyGate extends TestJDBCBackend {
TableMaintenanceActionMetaService actions =
TableMaintenanceActionMetaService.getInstance();
OccupancyGate gate = new OccupancyGate(actions);
- long policyId = 22L;
+ long policyId = insertPolicy(metalakeName, "occ-policy").id();
Assertions.assertFalse(gate.isInFlight(table.id(), policyId));
@@ -95,7 +100,7 @@ class TestOccupancyGate extends TestJDBCBackend {
.build());
Assertions.assertFalse(gate.isInFlight(table.id(), policyId));
- Assertions.assertEquals(0L, actions.getByJobId(started.id()).finishedAt());
+ Assertions.assertEquals(finishedAt,
actions.getByJobId(started.id()).finishedAt());
Assertions.assertTrue(
gate.withinMinInterval(table.id(), policyId, 3_600_000L, finishedAt +
1_000L));
Assertions.assertFalse(
@@ -127,7 +132,7 @@ class TestOccupancyGate extends TestJDBCBackend {
}
@TestTemplate
- void testSoftSlotBlocksNewAcquireUntilMarkFinished() throws IOException {
+ void testTerminalBoundJobIsReclaimedSoSlotCanBeAcquiredAgain() throws
IOException {
String metalakeName = "tms_occ_ml3";
createAndInsertMakeLake(metalakeName);
createAndInsertCatalog(metalakeName, "cat");
@@ -138,7 +143,7 @@ class TestOccupancyGate extends TestJDBCBackend {
TableMaintenanceActionMetaService actions =
TableMaintenanceActionMetaService.getInstance();
OccupancyGate gate = new OccupancyGate(actions);
- long policyId = 55L;
+ long policyId = insertPolicy(metalakeName, "occ-policy-3").id();
long finishedAt = 1_700_000_200_000L;
JobEntity succeeded =
@@ -159,13 +164,7 @@ class TestOccupancyGate extends TestJDBCBackend {
actions.bindJobId(stale, succeeded.id());
Assertions.assertEquals(0L,
actions.getByJobId(succeeded.id()).finishedAt());
Assertions.assertFalse(gate.isInFlight(table.id(), policyId));
- Assertions.assertThrows(
- IllegalStateException.class,
- () ->
- actions.insertInFlight(
- table.id(), policyId,
TableMaintenanceConfig.DEFAULT_CLAIM_TIMEOUT_MS));
-
- actions.markFinished(succeeded.id(), 200L, finishedAt);
+ Assertions.assertEquals(finishedAt,
actions.getByJobId(succeeded.id()).finishedAt());
Assertions.assertEquals(0, actions.listInFlight(table.id(),
policyId).size());
JobEntity started =
@@ -191,6 +190,143 @@ class TestOccupancyGate extends TestJDBCBackend {
gate.shouldSkip(table.id(), policyId, 3_600_000L, finishedAt +
3_600_001L));
}
+ @TestTemplate
+ void testMissingJobMetaIsReclaimedSoSlotCanBeAcquiredAgain() throws
IOException {
+ String metalakeName = "tms_occ_ml5";
+ createAndInsertMakeLake(metalakeName);
+ createAndInsertCatalog(metalakeName, "cat");
+ createAndInsertSchema(metalakeName, "cat", "sch");
+ TableEntity table = createAndInsertTableEntity(Namespace.of(metalakeName,
"cat", "sch"), "tbl");
+
+ TableMaintenanceActionMetaService actions =
TableMaintenanceActionMetaService.getInstance();
+ OccupancyGate gate = new OccupancyGate(actions);
+ // Policy id need not exist: job lookup goes through the table's metalake.
+ long policyId = 77L;
+ long orphanJobId = 7005L;
+
+ OccupancyClaim stale =
+ actions.insertInFlight(
+ table.id(), policyId,
TableMaintenanceConfig.DEFAULT_CLAIM_TIMEOUT_MS);
+ actions.bindJobId(stale, orphanJobId);
+ Assertions.assertEquals(0L, actions.getByJobId(orphanJobId).finishedAt());
+ // No JobEntity row for orphanJobId — reclaim must still free the soft
slot.
+ Assertions.assertFalse(gate.isInFlight(table.id(), policyId));
+ Assertions.assertTrue(actions.getByJobId(orphanJobId).finishedAt() > 0L);
+ Assertions.assertEquals(0, actions.listInFlight(table.id(),
policyId).size());
+ Assertions.assertTrue(
+ actions
+ .tryAcquireInFlight(
+ table.id(), policyId,
TableMaintenanceConfig.DEFAULT_CLAIM_TIMEOUT_MS)
+ .isPresent());
+ }
+
+ @TestTemplate
+ void testReclaimDoesNotCopyBeforeSnapshotIdIntoAfter() throws IOException {
+ String metalakeName = "tms_occ_ml6";
+ createAndInsertMakeLake(metalakeName);
+ createAndInsertCatalog(metalakeName, "cat");
+ createAndInsertSchema(metalakeName, "cat", "sch");
+ TableEntity table = createAndInsertTableEntity(Namespace.of(metalakeName,
"cat", "sch"), "tbl");
+ JobTemplateEntity template =
+ createAndInsertShellJobTemplateEntity("occ-template-6", "comment",
metalakeName);
+
+ TableMaintenanceActionMetaService actions =
TableMaintenanceActionMetaService.getInstance();
+ OccupancyGate gate = new OccupancyGate(actions);
+ long policyId = insertPolicy(metalakeName, "occ-policy-6").id();
+ long finishedAt = 1_700_000_300_000L;
+ long beforeSnapshotId = 100L;
+
+ JobEntity succeeded =
+ JobEntity.builder()
+ .withId(7006L)
+ .withJobExecutionId("exec-7006")
+ .withNamespace(NamespaceUtil.ofJob(metalakeName))
+ .withJobTemplateName(template.name())
+ .withStatus(JobHandle.Status.SUCCEEDED)
+ .withAuditInfo(AUDIT_INFO)
+ .withStartedAt(finishedAt - 1_000L)
+ .withFinishedAt(finishedAt)
+ .build();
+ backend.insert(succeeded, false);
+ OccupancyClaim claim =
+ actions.insertInFlight(
+ table.id(), policyId,
TableMaintenanceConfig.DEFAULT_CLAIM_TIMEOUT_MS);
+ actions.bindJobId(claim, succeeded.id());
+ actions.recordBeforeSnapshot(succeeded.id(), beforeSnapshotId);
+ Assertions.assertEquals(
+ beforeSnapshotId,
actions.getByJobId(succeeded.id()).beforeSnapshotId());
+
Assertions.assertNull(actions.getByJobId(succeeded.id()).afterSnapshotId());
+
+ Assertions.assertFalse(gate.isInFlight(table.id(), policyId));
+ Assertions.assertEquals(finishedAt,
actions.getByJobId(succeeded.id()).finishedAt());
+ Assertions.assertEquals(
+ beforeSnapshotId,
actions.getByJobId(succeeded.id()).beforeSnapshotId());
+ // Failsafe reclaim must not invent S1 from S0 (§5.2.3).
+
Assertions.assertNull(actions.getByJobId(succeeded.id()).afterSnapshotId());
+
+ // Late finish listener may still backfill a real S1 after reclaim closed
the soft slot.
+ long afterSnapshotId = 200L;
+ actions.markFinished(succeeded.id(), afterSnapshotId,
System.currentTimeMillis());
+ Assertions.assertEquals(afterSnapshotId,
actions.getByJobId(succeeded.id()).afterSnapshotId());
+ Assertions.assertEquals(finishedAt,
actions.getByJobId(succeeded.id()).finishedAt());
+ }
+
+ @TestTemplate
+ void testActiveJobNotReclaimedWhenPolicyIdIsUnknown() throws IOException {
+ String metalakeName = "tms_occ_ml7";
+ createAndInsertMakeLake(metalakeName);
+ createAndInsertCatalog(metalakeName, "cat");
+ createAndInsertSchema(metalakeName, "cat", "sch");
+ TableEntity table = createAndInsertTableEntity(Namespace.of(metalakeName,
"cat", "sch"), "tbl");
+ JobTemplateEntity template =
+ createAndInsertShellJobTemplateEntity("occ-template-7", "comment",
metalakeName);
+
+ TableMaintenanceActionMetaService actions =
TableMaintenanceActionMetaService.getInstance();
+ OccupancyGate gate = new OccupancyGate(actions);
+ // Simulate a deleted / unreadable policy: lookup must still use the table
metalake.
+ long unknownPolicyId = 9_000_007L;
+
+ JobEntity started =
+ JobEntity.builder()
+ .withId(7007L)
+ .withJobExecutionId("exec-7007")
+ .withNamespace(NamespaceUtil.ofJob(metalakeName))
+ .withJobTemplateName(template.name())
+ .withStatus(JobHandle.Status.STARTED)
+ .withAuditInfo(AUDIT_INFO)
+ .withStartedAt(System.currentTimeMillis())
+ .withFinishedAt(0L)
+ .build();
+ backend.insert(started, false);
+ OccupancyClaim claim =
+ actions.insertInFlight(
+ table.id(), unknownPolicyId,
TableMaintenanceConfig.DEFAULT_CLAIM_TIMEOUT_MS);
+ actions.bindJobId(claim, started.id());
+
+ Assertions.assertTrue(gate.isInFlight(table.id(), unknownPolicyId));
+ Assertions.assertEquals(0L, actions.getByJobId(started.id()).finishedAt());
+ Assertions.assertEquals(1, actions.listInFlight(table.id(),
unknownPolicyId).size());
+ }
+
+ @TestTemplate
+ void testUnresolvedTableMetalakeDoesNotReclaimBoundRow() throws IOException {
+ TableMaintenanceActionMetaService actions =
TableMaintenanceActionMetaService.getInstance();
+ OccupancyGate gate = new OccupancyGate(actions);
+ long missingTableId = 9_999_001L;
+ long policyId = 88L;
+ long orphanJobId = 7008L;
+
+ OccupancyClaim claim =
+ actions.insertInFlight(
+ missingTableId, policyId,
TableMaintenanceConfig.DEFAULT_CLAIM_TIMEOUT_MS);
+ actions.bindJobId(claim, orphanJobId);
+
+ // Table/metalake cannot be resolved — do not treat the job as
missing/terminal.
+ Assertions.assertTrue(gate.isInFlight(missingTableId, policyId));
+ Assertions.assertEquals(0L, actions.getByJobId(orphanJobId).finishedAt());
+ Assertions.assertEquals(1, actions.listInFlight(missingTableId,
policyId).size());
+ }
+
@TestTemplate
void testExpiredUnboundClaimIsNotInFlightOnDbClock() throws Exception {
String metalakeName = "tms_occ_ml4";
@@ -218,4 +354,13 @@ class TestOccupancyGate extends TestJDBCBackend {
actions.tryAcquireInFlight(table.id(), policyId,
claimTimeoutMs).isPresent());
Assertions.assertEquals(claim.id(), actions.getById(claim.id()).id());
}
+
+ private PolicyEntity insertPolicy(String metalakeName, String policyName)
throws IOException {
+ return createAndInsertPolicyEntity(
+ policyName,
+ "comment",
+ PolicyContents.custom(
+ ImmutableMap.of("k", "v"),
ImmutableSet.of(MetadataObject.Type.TABLE), null),
+ metalakeName);
+ }
}
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
index 5979eef5d8..accf37292a 100644
---
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
@@ -74,6 +74,32 @@ class TestTableMaintenanceActionMetaService extends
TestJDBCBackend {
"{\"mse\":2}", metricsService.getMetrics(10L, 20L,
200L).metricsValue());
}
+ @TestTemplate
+ void testMarkFinishedBackfillsAfterSnapshotWhenAlreadyClosed() {
+ TableMaintenanceActionMetaService actionService =
+ TableMaintenanceActionMetaService.getInstance();
+ long jobId = 1999L;
+ OccupancyClaim claim =
+ actionService.insertInFlight(
+ 12L, 22L, TableMaintenanceConfig.forTest(1, 1, 1, 500L, 1_000L, 4,
10));
+ actionService.bindJobId(claim, jobId);
+ actionService.recordBeforeSnapshot(jobId, 100L);
+
+ long reclaimFinishedAt = System.currentTimeMillis();
+ // Failsafe reclaim closes with null S1.
+ actionService.markFinished(jobId, null, reclaimFinishedAt);
+ Assertions.assertEquals(reclaimFinishedAt,
actionService.getByJobId(jobId).finishedAt());
+ Assertions.assertNull(actionService.getByJobId(jobId).afterSnapshotId());
+
+ // Late finish listener still records the real terminal snapshot.
+ actionService.markFinished(jobId, 200L, System.currentTimeMillis());
+ Assertions.assertEquals(200L,
actionService.getByJobId(jobId).afterSnapshotId());
+ Assertions.assertEquals(reclaimFinishedAt,
actionService.getByJobId(jobId).finishedAt());
+ Assertions.assertThrows(
+ IllegalStateException.class,
+ () -> actionService.markFinished(jobId, 201L,
System.currentTimeMillis()));
+ }
+
@TestTemplate
void testTryAcquireRejectsLiveUnboundAndReclaimsExpired() throws Exception {
TableMaintenanceActionMetaService actions =
TableMaintenanceActionMetaService.getInstance();
