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 f356e785fea5016005aa3fb3ef56b243dd073b0c Author: remilu <[email protected]> AuthorDate: Mon Oct 26 20:31:49 2026 +0800 [#13760] fix(policy): Reconcile snapshot-expiration thresholds with execution peers Adopt main's olderThanDays/retainLast/streamResults (DTO/SerDe/tests) while keeping PolicyExecutionSpec peers. Also keeps the rewrite-manifests factory adaptation for the combined thresholds+peers API. Drop the obsolete expire trigger-expr=false placeholder unit test; production routes expire to AlwaysMaintain, and content now ships trigger-expr=true. Keep coverage for missing trigger-expr → SKIP. Co-authored-by: Cursor <[email protected]> --- .../policy/IcebergSnapshotExpirationContent.java | 36 +++-- .../apache/gravitino/policy/PolicyContents.java | 57 +++++++- .../gravitino/policy/TestPolicyContents.java | 56 ++++---- .../gravitino/dto/policy/PolicyContentDTO.java | 97 ++++++++++---- .../apache/gravitino/dto/util/DTOConverters.java | 7 +- .../json/ExecutablePolicyContentSerde.java | 54 ++++++-- .../apache/gravitino/dto/policy/TestPolicyDTO.java | 3 +- ...iteManifestsAndSnapshotExpirationPolicyDTO.java | 5 + .../storage/relational/utils/TestPOConverters.java | 147 +++++++++++++++++++-- .../injob/AlwaysMaintainInJobRecommender.java | 6 +- .../injob/TriggerExprInJobRecommender.java | 5 +- .../injob/TestTriggerExprInJobRecommender.java | 14 +- 12 files changed, 370 insertions(+), 117 deletions(-) diff --git a/api/src/main/java/org/apache/gravitino/policy/IcebergSnapshotExpirationContent.java b/api/src/main/java/org/apache/gravitino/policy/IcebergSnapshotExpirationContent.java index a655b93af1..d2cc6fd5ed 100644 --- a/api/src/main/java/org/apache/gravitino/policy/IcebergSnapshotExpirationContent.java +++ b/api/src/main/java/org/apache/gravitino/policy/IcebergSnapshotExpirationContent.java @@ -77,6 +77,15 @@ public class IcebergSnapshotExpirationContent extends AbstractExecutablePolicyCo IcebergSnapshotExpirationContent( @Nullable Long olderThanDays, @Nullable Long retainLast, @Nullable Boolean streamResults) { + this(olderThanDays, retainLast, streamResults, PolicyExecutionSpec.empty()); + } + + IcebergSnapshotExpirationContent( + @Nullable Long olderThanDays, + @Nullable Long retainLast, + @Nullable Boolean streamResults, + @Nullable PolicyExecutionSpec executionSpec) { + super(executionSpec); this.olderThanDays = olderThanDays == null ? DEFAULT_OLDER_THAN_DAYS : olderThanDays; this.retainLast = retainLast == null ? DEFAULT_RETAIN_LAST : retainLast; this.streamResults = streamResults == null ? DEFAULT_STREAM_RESULTS : streamResults; @@ -152,7 +161,7 @@ public class IcebergSnapshotExpirationContent extends AbstractExecutablePolicyCo @Override public void validate() { - PolicyContent.super.validate(); + super.validate(); validateOlderThanDays(olderThanDays); validateRetainLast(retainLast); } @@ -163,25 +172,30 @@ public class IcebergSnapshotExpirationContent extends AbstractExecutablePolicyCo return false; } IcebergSnapshotExpirationContent that = (IcebergSnapshotExpirationContent) other; - return olderThanDays == that.olderThanDays + return executionFieldsEqual(that) + && olderThanDays == that.olderThanDays && retainLast == that.retainLast && streamResults == that.streamResults; } @Override public int hashCode() { - return Objects.hash(olderThanDays, retainLast, streamResults); + return Objects.hash(executionFieldsHashCode(), olderThanDays, retainLast, streamResults); } @Override public String toString() { - return "IcebergSnapshotExpirationContent{" - + "olderThanDays=" - + olderThanDays - + ", retainLast=" - + retainLast - + ", streamResults=" - + streamResults - + '}'; + StringBuilder sb = + new StringBuilder("IcebergSnapshotExpirationContent{") + .append("olderThanDays=") + .append(olderThanDays) + .append(", retainLast=") + .append(retainLast) + .append(", streamResults=") + .append(streamResults) + .append(", "); + appendExecutionFields(sb); + sb.append('}'); + return sb.toString(); } } diff --git a/api/src/main/java/org/apache/gravitino/policy/PolicyContents.java b/api/src/main/java/org/apache/gravitino/policy/PolicyContents.java index df994c4b0c..0c354d53d5 100644 --- a/api/src/main/java/org/apache/gravitino/policy/PolicyContents.java +++ b/api/src/main/java/org/apache/gravitino/policy/PolicyContents.java @@ -370,7 +370,62 @@ public class PolicyContents { */ public static IcebergSnapshotExpirationContent icebergSnapshotExpiration( @Nullable Long olderThanDays, @Nullable Long retainLast, @Nullable Boolean streamResults) { - return new IcebergSnapshotExpirationContent(olderThanDays, retainLast, streamResults); + return icebergSnapshotExpiration( + olderThanDays, retainLast, streamResults, null, null, null, null, null); + } + + /** + * Creates Iceberg snapshot-expiration policy content with default thresholds and execution peers. + * + * <p>Prefer this over mixing peers into the thresholds overload so {@code (Long, Long, Boolean)} + * stays unambiguous. + * + * @param timeTrigger optional cron expression + * @param commitTrigger optional commit-trigger flag + * @param minIntervalMs optional cooldown ms + * @param jobOptions optional job options + * @param updaterOptions optional updater options + * @return snapshot-expiration policy content + */ + public static PolicyContent icebergSnapshotExpiration( + @Nullable String timeTrigger, + @Nullable Boolean commitTrigger, + @Nullable Long minIntervalMs, + @Nullable Map<String, String> jobOptions, + @Nullable Map<String, String> updaterOptions) { + return icebergSnapshotExpiration( + null, null, null, timeTrigger, commitTrigger, minIntervalMs, jobOptions, updaterOptions); + } + + /** + * Creates Iceberg snapshot-expiration policy content with thresholds and execution peers. Null + * thresholds use built-in defaults. + * + * @param olderThanDays minimum snapshot age in days, or null for the default + * @param retainLast minimum recent snapshots to keep, or null for the default + * @param streamResults whether to stream intermediate delete results, or null for the default + * @param timeTrigger optional cron expression + * @param commitTrigger optional commit-trigger flag + * @param minIntervalMs optional cooldown ms + * @param jobOptions optional job options + * @param updaterOptions optional updater options + * @return snapshot-expiration policy content + */ + public static IcebergSnapshotExpirationContent icebergSnapshotExpiration( + @Nullable Long olderThanDays, + @Nullable Long retainLast, + @Nullable Boolean streamResults, + @Nullable String timeTrigger, + @Nullable Boolean commitTrigger, + @Nullable Long minIntervalMs, + @Nullable Map<String, String> jobOptions, + @Nullable Map<String, String> updaterOptions) { + return new IcebergSnapshotExpirationContent( + olderThanDays, + retainLast, + streamResults, + PolicyExecutionSpec.of( + timeTrigger, commitTrigger, minIntervalMs, jobOptions, updaterOptions)); } /** diff --git a/api/src/test/java/org/apache/gravitino/policy/TestPolicyContents.java b/api/src/test/java/org/apache/gravitino/policy/TestPolicyContents.java index bcd8e3a780..9d30954b38 100644 --- a/api/src/test/java/org/apache/gravitino/policy/TestPolicyContents.java +++ b/api/src/test/java/org/apache/gravitino/policy/TestPolicyContents.java @@ -180,6 +180,34 @@ public class TestPolicyContents { Assertions.assertTrue(exception.getMessage().contains("maxPartitionNum")); } + @Test + void testIcebergCompactionContentRejectsUnknownOrInvalidRewriteOptions() { + IllegalArgumentException unknown = + Assertions.assertThrows( + IllegalArgumentException.class, + () -> + PolicyContents.icebergDataCompaction(1000L, 1L, mapOf("no-such-option", "x")) + .validate()); + Assertions.assertTrue(unknown.getMessage().contains("no-such-option")); + + IllegalArgumentException negative = + Assertions.assertThrows( + IllegalArgumentException.class, + () -> + PolicyContents.icebergDataCompaction( + 1000L, 1L, mapOf("target-file-size-bytes", "-1")) + .validate()); + Assertions.assertTrue(negative.getMessage().contains("target-file-size-bytes")); + + IllegalArgumentException notBool = + Assertions.assertThrows( + IllegalArgumentException.class, + () -> + PolicyContents.icebergDataCompaction(1000L, 1L, mapOf("rewrite-all", "yes")) + .validate()); + Assertions.assertTrue(notBool.getMessage().contains("rewrite-all")); + } + @Test void testOfRejectsNegativeMinIntervalMsOnExecutableContent() { IllegalArgumentException exception = @@ -244,34 +272,6 @@ public class TestPolicyContents { Assertions.assertTrue(exception.getMessage().contains("commitTrigger")); } - @Test - void testIcebergCompactionContentRejectsUnknownOrInvalidRewriteOptions() { - IllegalArgumentException unknown = - Assertions.assertThrows( - IllegalArgumentException.class, - () -> - PolicyContents.icebergDataCompaction(1000L, 1L, mapOf("no-such-option", "x")) - .validate()); - Assertions.assertTrue(unknown.getMessage().contains("no-such-option")); - - IllegalArgumentException negative = - Assertions.assertThrows( - IllegalArgumentException.class, - () -> - PolicyContents.icebergDataCompaction( - 1000L, 1L, mapOf("target-file-size-bytes", "-1")) - .validate()); - Assertions.assertTrue(negative.getMessage().contains("target-file-size-bytes")); - - IllegalArgumentException notBool = - Assertions.assertThrows( - IllegalArgumentException.class, - () -> - PolicyContents.icebergDataCompaction(1000L, 1L, mapOf("rewrite-all", "yes")) - .validate()); - Assertions.assertTrue(notBool.getMessage().contains("rewrite-all")); - } - @Test void testAccessControlContent() { AccessControlContent content = diff --git a/common/src/main/java/org/apache/gravitino/dto/policy/PolicyContentDTO.java b/common/src/main/java/org/apache/gravitino/dto/policy/PolicyContentDTO.java index 7e10253286..24562a9cb0 100644 --- a/common/src/main/java/org/apache/gravitino/dto/policy/PolicyContentDTO.java +++ b/common/src/main/java/org/apache/gravitino/dto/policy/PolicyContentDTO.java @@ -845,10 +845,10 @@ public interface PolicyContentDTO extends PolicyContent { * defaults rather than merging with the previous content. */ @EqualsAndHashCode - @ToString + @ToString(exclude = {"jobOptions", "updaterOptions"}) @Builder(setterPrefix = "with") @AllArgsConstructor(access = AccessLevel.PRIVATE) - class IcebergSnapshotExpirationContentDTO implements PolicyContentDTO { + class IcebergSnapshotExpirationContentDTO implements PolicyContentDTO, ExecutionPeersDTO { @JsonProperty("olderThanDays") @JsonDeserialize(using = SnapshotLongDeserializer.class) private Long olderThanDays; @@ -861,6 +861,34 @@ public interface PolicyContentDTO extends PolicyContent { @JsonDeserialize(using = SnapshotBooleanDeserializer.class) private Boolean streamResults; + @EqualsAndHashCode.Exclude + @Nullable + @JsonProperty("timeTrigger") + @JsonInclude(JsonInclude.Include.NON_NULL) + private String timeTrigger; + + @EqualsAndHashCode.Exclude + @Nullable + @JsonProperty("commitTrigger") + @JsonInclude(JsonInclude.Include.NON_NULL) + private Boolean commitTrigger; + + @EqualsAndHashCode.Exclude + @Nullable + @JsonProperty("minIntervalMs") + @JsonInclude(JsonInclude.Include.NON_NULL) + private Long minIntervalMs; + + @EqualsAndHashCode.Exclude + @JsonProperty("jobOptions") + @JsonInclude(JsonInclude.Include.NON_EMPTY) + private Map<String, String> jobOptions; + + @EqualsAndHashCode.Exclude + @JsonProperty("updaterOptions") + @JsonInclude(JsonInclude.Include.NON_EMPTY) + private Map<String, String> updaterOptions; + private IcebergSnapshotExpirationContentDTO() {} /** @@ -943,34 +971,20 @@ public interface PolicyContentDTO extends PolicyContent { toDomainContent().validate(); } - /** Reads a long option without silently coercing fractional values or strings. */ - public static class SnapshotLongDeserializer extends JsonDeserializer<Long> { - @Override - public Long deserialize(JsonParser parser, DeserializationContext context) - throws IOException { - if (!parser.hasToken(JsonToken.VALUE_NUMBER_INT)) { - return context.reportInputMismatch( - Long.class, "olderThanDays and retainLast must be 64-bit integers"); - } - return parser.getLongValue(); - } - } - - /** Reads streamResults without silently coercing numbers or strings to booleans. */ - public static class SnapshotBooleanDeserializer extends JsonDeserializer<Boolean> { - @Override - public Boolean deserialize(JsonParser parser, DeserializationContext context) - throws IOException { - if (!parser.hasToken(JsonToken.VALUE_TRUE) && !parser.hasToken(JsonToken.VALUE_FALSE)) { - return context.reportInputMismatch(Boolean.class, "streamResults must be a boolean"); - } - return parser.getBooleanValue(); - } - } - - private PolicyContent toDomainContent() { + /** + * @return domain policy content including execution peers + */ + public PolicyContent toDomainContent() { + PolicyExecutionSpec spec = executionSpec(); return PolicyContents.icebergSnapshotExpiration( - olderThanDays(), retainLast(), streamResults()); + olderThanDays(), + retainLast(), + streamResults(), + spec.timeTrigger(), + spec.commitTrigger(), + spec.minIntervalMs(), + spec.jobOptions(), + spec.updaterOptions()); } /** Lombok builder customization: unpack PolicyExecutionSpec into flat peer fields. */ @@ -998,6 +1012,31 @@ public interface PolicyContentDTO extends PolicyContent { return this; } } + + /** Reads a long option without silently coercing fractional values or strings. */ + public static class SnapshotLongDeserializer extends JsonDeserializer<Long> { + @Override + public Long deserialize(JsonParser parser, DeserializationContext context) + throws IOException { + if (!parser.hasToken(JsonToken.VALUE_NUMBER_INT)) { + return context.reportInputMismatch( + Long.class, "olderThanDays and retainLast must be 64-bit integers"); + } + return parser.getLongValue(); + } + } + + /** Reads streamResults without silently coercing numbers or strings to booleans. */ + public static class SnapshotBooleanDeserializer extends JsonDeserializer<Boolean> { + @Override + public Boolean deserialize(JsonParser parser, DeserializationContext context) + throws IOException { + if (!parser.hasToken(JsonToken.VALUE_TRUE) && !parser.hasToken(JsonToken.VALUE_FALSE)) { + return context.reportInputMismatch(Boolean.class, "streamResults must be a boolean"); + } + return parser.getBooleanValue(); + } + } } /** Represents typed row-filter policy content. */ diff --git a/common/src/main/java/org/apache/gravitino/dto/util/DTOConverters.java b/common/src/main/java/org/apache/gravitino/dto/util/DTOConverters.java index 19e0072c87..5fdf3c51f4 100644 --- a/common/src/main/java/org/apache/gravitino/dto/util/DTOConverters.java +++ b/common/src/main/java/org/apache/gravitino/dto/util/DTOConverters.java @@ -715,6 +715,7 @@ public class DTOConverters { .withOlderThanDays(content.olderThanDays()) .withRetainLast(content.retainLast()) .withStreamResults(content.streamResults()) + .withExecutionSpec(executionSpec) .build(); } if (policyContent instanceof IcebergDataCompactionContent) { @@ -1608,10 +1609,8 @@ public class DTOConverters { .toDomainContent(); } if (policyContentDTO instanceof PolicyContentDTO.IcebergSnapshotExpirationContentDTO) { - PolicyContentDTO.IcebergSnapshotExpirationContentDTO content = - (PolicyContentDTO.IcebergSnapshotExpirationContentDTO) policyContentDTO; - return PolicyContents.icebergSnapshotExpiration( - content.olderThanDays(), content.retainLast(), content.streamResults()); + return ((PolicyContentDTO.IcebergSnapshotExpirationContentDTO) policyContentDTO) + .toDomainContent(); } if (policyContentDTO instanceof PolicyContentDTO.IcebergCompactionContentDTO) { return ((PolicyContentDTO.IcebergCompactionContentDTO) policyContentDTO).toDomainContent(); diff --git a/common/src/main/java/org/apache/gravitino/json/ExecutablePolicyContentSerde.java b/common/src/main/java/org/apache/gravitino/json/ExecutablePolicyContentSerde.java index a033614e57..3dc1d1aa86 100644 --- a/common/src/main/java/org/apache/gravitino/json/ExecutablePolicyContentSerde.java +++ b/common/src/main/java/org/apache/gravitino/json/ExecutablePolicyContentSerde.java @@ -267,6 +267,15 @@ public final class ExecutablePolicyContentSerde { IcebergRewriteManifestsContent value, JsonGenerator gen, SerializerProvider serializers) throws IOException { gen.writeStartObject(); + gen.writeNumberField("manifestCountCritical", value.manifestCountCritical()); + gen.writeNumberField("manifestCountWarning", value.manifestCountWarning()); + gen.writeNumberField("avgManifestSizeThresholdBytes", value.avgManifestSizeThresholdBytes()); + if (value.specId() != null) { + gen.writeNumberField("specId", value.specId()); + } + if (value.useCaching() != null) { + gen.writeBooleanField("useCaching", value.useCaching()); + } writeExecutionPeerFields(value.executionSpec(), gen); gen.writeEndObject(); } @@ -283,13 +292,17 @@ public final class ExecutablePolicyContentSerde { node != null && node.isObject(), "Cannot parse rewrite-manifests content from invalid JSON: %s", node); - return (IcebergRewriteManifestsContent) - PolicyContents.icebergRewriteManifests( - textOrNull(node, "timeTrigger"), - booleanOrNull(node, "commitTrigger"), - longOrNull(node, "minIntervalMs"), - stringMapOrNull(node, "jobOptions"), - stringMapOrNull(node, "updaterOptions")); + return PolicyContents.icebergRewriteManifests( + longOrNull(node, "manifestCountCritical"), + longOrNull(node, "manifestCountWarning"), + longOrNull(node, "avgManifestSizeThresholdBytes"), + integerOrNull(node, "specId"), + booleanOrNull(node, "useCaching"), + textOrNull(node, "timeTrigger"), + booleanOrNull(node, "commitTrigger"), + longOrNull(node, "minIntervalMs"), + stringMapOrNull(node, "jobOptions"), + stringMapOrNull(node, "updaterOptions")); } } @@ -301,6 +314,9 @@ public final class ExecutablePolicyContentSerde { IcebergSnapshotExpirationContent value, JsonGenerator gen, SerializerProvider serializers) throws IOException { gen.writeStartObject(); + gen.writeNumberField("olderThanDays", value.olderThanDays()); + gen.writeNumberField("retainLast", value.retainLast()); + gen.writeBooleanField("streamResults", value.streamResults()); writeExecutionPeerFields(value.executionSpec(), gen); gen.writeEndObject(); } @@ -317,13 +333,15 @@ public final class ExecutablePolicyContentSerde { node != null && node.isObject(), "Cannot parse snapshot-expiration content from invalid JSON: %s", node); - return (IcebergSnapshotExpirationContent) - PolicyContents.icebergSnapshotExpiration( - textOrNull(node, "timeTrigger"), - booleanOrNull(node, "commitTrigger"), - longOrNull(node, "minIntervalMs"), - stringMapOrNull(node, "jobOptions"), - stringMapOrNull(node, "updaterOptions")); + return PolicyContents.icebergSnapshotExpiration( + longOrNull(node, "olderThanDays"), + longOrNull(node, "retainLast"), + booleanOrNull(node, "streamResults"), + textOrNull(node, "timeTrigger"), + booleanOrNull(node, "commitTrigger"), + longOrNull(node, "minIntervalMs"), + stringMapOrNull(node, "jobOptions"), + stringMapOrNull(node, "updaterOptions")); } } @@ -351,6 +369,14 @@ public final class ExecutablePolicyContentSerde { return node.get(property).asLong(); } + @Nullable + private static Integer integerOrNull(JsonNode node, String property) { + if (!node.has(property) || node.get(property).isNull()) { + return null; + } + return node.get(property).asInt(); + } + @Nullable private static Map<String, String> stringMapOrNull(JsonNode node, String property) { if (!node.has(property) || node.get(property).isNull()) { diff --git a/common/src/test/java/org/apache/gravitino/dto/policy/TestPolicyDTO.java b/common/src/test/java/org/apache/gravitino/dto/policy/TestPolicyDTO.java index 666f46975b..042c301819 100644 --- a/common/src/test/java/org/apache/gravitino/dto/policy/TestPolicyDTO.java +++ b/common/src/test/java/org/apache/gravitino/dto/policy/TestPolicyDTO.java @@ -29,9 +29,9 @@ import org.apache.gravitino.authorization.Privilege; import org.apache.gravitino.dto.AuditDTO; import org.apache.gravitino.json.JsonUtils; import org.apache.gravitino.policy.IcebergDataCompactionContent; +import org.apache.gravitino.policy.PolicyExecutionSpec; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; -import org.apache.gravitino.policy.PolicyExecutionSpec; public class TestPolicyDTO { @@ -321,5 +321,4 @@ public class TestPolicyDTO { .build(); Assertions.assertDoesNotThrow(opaqueCron::validate); } - } diff --git a/common/src/test/java/org/apache/gravitino/dto/policy/TestRewriteManifestsAndSnapshotExpirationPolicyDTO.java b/common/src/test/java/org/apache/gravitino/dto/policy/TestRewriteManifestsAndSnapshotExpirationPolicyDTO.java index 3ce7b9ea05..0cdd803469 100644 --- a/common/src/test/java/org/apache/gravitino/dto/policy/TestRewriteManifestsAndSnapshotExpirationPolicyDTO.java +++ b/common/src/test/java/org/apache/gravitino/dto/policy/TestRewriteManifestsAndSnapshotExpirationPolicyDTO.java @@ -49,6 +49,11 @@ class TestRewriteManifestsAndSnapshotExpirationPolicyDTO { assertResponseRoundTripWithPeers( "system_iceberg_rewrite_manifests", PolicyContents.icebergRewriteManifests( + null, + null, + null, + null, + null, "0 30 1 * * ?", true, 900_000L, diff --git a/core/src/test/java/org/apache/gravitino/storage/relational/utils/TestPOConverters.java b/core/src/test/java/org/apache/gravitino/storage/relational/utils/TestPOConverters.java index 464485f111..1a22aa0600 100644 --- a/core/src/test/java/org/apache/gravitino/storage/relational/utils/TestPOConverters.java +++ b/core/src/test/java/org/apache/gravitino/storage/relational/utils/TestPOConverters.java @@ -1035,6 +1035,112 @@ public class TestPOConverters { assertEquals(expectedPolicy.content(), convertedPolicy.content()); } + @Test + public void testFromPolicyPOIcebergCompactionWithExecutionPeers() throws JsonProcessingException { + PolicyContent content = + PolicyContents.icebergDataCompaction( + IcebergDataCompactionContent.DEFAULT_MIN_DATA_FILE_MSE, + IcebergDataCompactionContent.DEFAULT_MIN_DELETE_FILE_NUMBER, + IcebergDataCompactionContent.DEFAULT_DATA_FILE_MSE_WEIGHT, + IcebergDataCompactionContent.DEFAULT_DELETE_FILE_NUMBER_WEIGHT, + IcebergDataCompactionContent.DEFAULT_MAX_PARTITION_NUM, + IcebergDataCompactionContent.DEFAULT_REWRITE_STRATEGY, + IcebergDataCompactionContent.DEFAULT_SORT_ORDER, + ImmutableMap.of("target-file-size-bytes", "134217728"), + "0 0 2 * * ?", + true, + 3_600_000L, + ImmutableMap.of("spark.executor.memory", "4g"), + ImmutableMap.of("metalake", "ml")); + + // Persistence must keep peer keys flat (not nested under executionSpec). + String storedJson = JsonUtils.anyFieldMapper().writeValueAsString(content); + assertTrue(storedJson.contains("\"timeTrigger\"")); + assertTrue(storedJson.contains("\"commitTrigger\"")); + assertTrue(storedJson.contains("\"minIntervalMs\"")); + assertTrue(storedJson.contains("\"jobOptions\"")); + assertTrue(storedJson.contains("\"updaterOptions\"")); + assertFalse(storedJson.contains("\"executionSpec\"")); + + PolicyVersionPO policyVersionPO = + createPolicyVersionPO(1L, 1L, 1L, "compaction with peers", true, content); + PolicyPO policyPO = + createPolicyPO(1L, "iceberg-compaction", "system_iceberg_compaction", 1L, policyVersionPO); + + PolicyEntity convertedPolicy = + POConverters.fromPolicyPO(policyPO, NamespaceUtil.ofPolicy("test_metalake")); + + assertEquals(Policy.BuiltInType.ICEBERG_COMPACTION, convertedPolicy.policyType()); + assertEquals(content, convertedPolicy.content()); + + IcebergDataCompactionContent loaded = (IcebergDataCompactionContent) convertedPolicy.content(); + PolicyExecutionSpec peers = loaded.executionSpec(); + assertEquals("0 0 2 * * ?", peers.timeTrigger()); + assertEquals(Boolean.TRUE, peers.commitTrigger()); + assertEquals(3_600_000L, peers.minIntervalMs()); + assertEquals("4g", peers.jobOptions().get("spark.executor.memory")); + assertEquals("ml", peers.updaterOptions().get("metalake")); + assertEquals("134217728", loaded.rewriteOptions().get("target-file-size-bytes")); + } + + @Test + public void testFromPolicyPOIcebergOrphanFileRemovalWithExecutionPeers() + throws JsonProcessingException { + PolicyContent content = + PolicyContents.icebergOrphanFileRemoval( + 7L, + "s3://bucket/table/data", + true, + "0 0 3 * * ?", + null, + 1_800_000L, + ImmutableMap.of("spark.executor.cores", "2"), + ImmutableMap.of("metalake", "ml")); + assertFlatExecutionPeersPersisted(content); + + PolicyEntity converted = + POConverters.fromPolicyPO( + createPolicyPO( + 2L, + "orphan-cleanup", + "system_iceberg_orphan_file_removal", + 1L, + createPolicyVersionPO(2L, 1L, 1L, "orphan peers", true, content)), + NamespaceUtil.ofPolicy("test_metalake")); + assertEquals(Policy.BuiltInType.ICEBERG_ORPHAN_FILE_REMOVAL, converted.policyType()); + assertEquals(content, converted.content()); + } + + @Test + public void testFromPolicyPOIcebergRewriteManifestsWithExecutionPeers() + throws JsonProcessingException { + PolicyContent content = + PolicyContents.icebergRewriteManifests( + null, + null, + null, + null, + null, + "0 30 1 * * ?", + true, + 900_000L, + ImmutableMap.of("spark.executor.memory", "2g"), + ImmutableMap.of("queue", "tms")); + assertFlatExecutionPeersPersisted(content); + + PolicyEntity converted = + POConverters.fromPolicyPO( + createPolicyPO( + 3L, + "rewrite-manifests", + "system_iceberg_rewrite_manifests", + 1L, + createPolicyVersionPO(3L, 1L, 1L, "rewrite peers", true, content)), + NamespaceUtil.ofPolicy("test_metalake")); + assertEquals(Policy.BuiltInType.ICEBERG_REWRITE_MANIFESTS, converted.policyType()); + assertEquals(content, converted.content()); + } + @Test public void testFromPolicyPOIcebergSnapshotExpiration() throws JsonProcessingException { PolicyContent content = PolicyContents.icebergSnapshotExpiration(14L, 5L, true); @@ -1046,35 +1152,54 @@ public class TestPOConverters { PolicyEntity converted = POConverters.fromPolicyPO( createPolicyPO( - 2L, + 5L, "snapshot-expiration", "system_iceberg_snapshot_expiration", 1L, - createPolicyVersionPO(2L, 1L, 1L, "expire", true, content)), + createPolicyVersionPO(5L, 1L, 1L, "expire", true, content)), NamespaceUtil.ofPolicy("test_metalake")); assertEquals(Policy.BuiltInType.ICEBERG_SNAPSHOT_EXPIRATION, converted.policyType()); assertEquals(content, converted.content()); } @Test - public void testFromPolicyPOIcebergRewriteManifestsPersists() throws JsonProcessingException { - PolicyContent content = PolicyContents.icebergRewriteManifests(); - String stored = JsonUtils.anyFieldMapper().writeValueAsString(content); - assertFalse(stored.isEmpty()); + public void testFromPolicyPOIcebergSnapshotExpirationWithExecutionPeers() + throws JsonProcessingException { + PolicyContent content = + PolicyContents.icebergSnapshotExpiration( + 14L, + 5L, + true, + "0 0 4 * * ?", + false, + 2_700_000L, + ImmutableMap.of("spark.driver.memory", "1g"), + ImmutableMap.of("metalake", "ml")); + assertFlatExecutionPeersPersisted(content); PolicyEntity converted = POConverters.fromPolicyPO( createPolicyPO( - 3L, - "rewrite-manifests", - "system_iceberg_rewrite_manifests", + 4L, + "snapshot-expiration", + "system_iceberg_snapshot_expiration", 1L, - createPolicyVersionPO(3L, 1L, 1L, "rewrite", true, content)), + createPolicyVersionPO(4L, 1L, 1L, "snapshot peers", true, content)), NamespaceUtil.ofPolicy("test_metalake")); - assertEquals(Policy.BuiltInType.ICEBERG_REWRITE_MANIFESTS, converted.policyType()); + assertEquals(Policy.BuiltInType.ICEBERG_SNAPSHOT_EXPIRATION, converted.policyType()); assertEquals(content, converted.content()); } + private static void assertFlatExecutionPeersPersisted(PolicyContent content) + throws JsonProcessingException { + String storedJson = JsonUtils.anyFieldMapper().writeValueAsString(content); + assertTrue(storedJson.contains("\"timeTrigger\"")); + assertTrue(storedJson.contains("\"minIntervalMs\"")); + assertTrue(storedJson.contains("\"jobOptions\"")); + assertTrue(storedJson.contains("\"updaterOptions\"")); + assertFalse(storedJson.contains("\"executionSpec\"")); + } + @Test public void testFromTagPO() throws JsonProcessingException { TagPO tagPO = createTagPO(1L, "test", 1L, "this is test"); diff --git a/plugins/maintenance/src/main/java/org/apache/gravitino/maintenance/injob/AlwaysMaintainInJobRecommender.java b/plugins/maintenance/src/main/java/org/apache/gravitino/maintenance/injob/AlwaysMaintainInJobRecommender.java index 188bd7643c..9c15d273be 100644 --- a/plugins/maintenance/src/main/java/org/apache/gravitino/maintenance/injob/AlwaysMaintainInJobRecommender.java +++ b/plugins/maintenance/src/main/java/org/apache/gravitino/maintenance/injob/AlwaysMaintainInJobRecommender.java @@ -21,9 +21,9 @@ package org.apache.gravitino.maintenance.injob; /** * In-job recommender that always returns {@link RecommendDecision#MAINTAIN}. * - * <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. + * <p>Production {@link TypeDispatchingInJobRecommender} uses this for expire / rewrite-manifests: + * cron already scheduled the work, so the in-job gate always maintains. Also available for tests + * and injectable overrides. */ public final class AlwaysMaintainInJobRecommender implements InJobRecommender { diff --git a/plugins/maintenance/src/main/java/org/apache/gravitino/maintenance/injob/TriggerExprInJobRecommender.java b/plugins/maintenance/src/main/java/org/apache/gravitino/maintenance/injob/TriggerExprInJobRecommender.java index a5ee5a28e8..2f03ede0a8 100644 --- a/plugins/maintenance/src/main/java/org/apache/gravitino/maintenance/injob/TriggerExprInJobRecommender.java +++ b/plugins/maintenance/src/main/java/org/apache/gravitino/maintenance/injob/TriggerExprInJobRecommender.java @@ -35,8 +35,9 @@ 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. + * trigger-expr (for example orphan cleanup {@code true}). Missing {@code trigger-expr} defaults to + * {@code false} ({@link RecommendDecision#SKIP}). Expire / rewrite-manifests are dispatched to + * {@link AlwaysMaintainInJobRecommender} instead. */ public final class TriggerExprInJobRecommender implements InJobRecommender { diff --git a/plugins/maintenance/src/test/java/org/apache/gravitino/maintenance/injob/TestTriggerExprInJobRecommender.java b/plugins/maintenance/src/test/java/org/apache/gravitino/maintenance/injob/TestTriggerExprInJobRecommender.java index bbf211564e..34a4019375 100644 --- a/plugins/maintenance/src/test/java/org/apache/gravitino/maintenance/injob/TestTriggerExprInJobRecommender.java +++ b/plugins/maintenance/src/test/java/org/apache/gravitino/maintenance/injob/TestTriggerExprInJobRecommender.java @@ -42,18 +42,8 @@ class TestTriggerExprInJobRecommender { } @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() { + void testMissingTriggerExprDefaultsToSkip() { + // rewrite-manifests rules omit trigger-expr; recommender defaults the expression to false. PolicyEntity policy = mock(PolicyEntity.class); when(policy.content()).thenReturn(PolicyContents.icebergRewriteManifests());
