This is an automated email from the ASF dual-hosted git repository. roryqi pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/gravitino.git
commit 88c3f0d12e2b99f3bb970a830376db238010310c Author: MaSai <[email protected]> AuthorDate: Thu Oct 22 09:08:42 2026 +0800 [#14065] feat(tms): In-job recommenders and terminal occupancy reclaim (#14054) 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. 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 No public REST. Occupancy reclaim changes for a maintenance action whose bound job is already terminal. `./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 | 28 ++++ 32 files changed, 1476 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 0352388420..1479343efd 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 @@ -20,8 +20,11 @@ package org.apache.gravitino.maintenance.occupancy; import static org.apache.gravitino.maintenance.storage.TableMaintenanceActionTestUtils.expireClaim; +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; @@ -29,7 +32,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; @@ -40,7 +45,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"); @@ -51,7 +56,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)); @@ -97,7 +102,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( @@ -129,7 +134,7 @@ class TestOccupancyGate extends TestJDBCBackend { } @TestTemplate - void testSoftSlotBlocksNewAcquireUntilMarkFinished() throws IOException { + void testTerminalBoundJobIsReclaimedSoSlotCanBeAcquiredAgain() throws IOException { String metalakeName = "tms_occ_ml3"; createAndInsertMakeLake(metalakeName); createAndInsertCatalog(metalakeName, "cat"); @@ -140,7 +145,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 = @@ -161,13 +166,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 = @@ -193,6 +192,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"; @@ -220,4 +356,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 e1b3d0c5a8..7c66eb4ad5 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 @@ -76,6 +76,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(); @@ -89,6 +115,8 @@ class TestTableMaintenanceActionMetaService extends TestJDBCBackend { Assertions.assertTrue(actions.tryAcquireInFlight(tableId, policyId, timeoutMs).isEmpty()); OccupancyClaim expired = expireClaim(first.get(), timeoutMs); + // Let the DB clock advance so reclaim cannot reuse the original fencing token. + Thread.sleep(2L); Optional<OccupancyClaim> reclaimed = actions.tryAcquireInFlight(tableId, policyId, timeoutMs); Assertions.assertTrue(reclaimed.isPresent());
