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 6869fbb797976d60a1a0473f7e0b6a172772d21e Author: MaSai <[email protected]> AuthorDate: Fri Oct 16 17:19:12 2026 +0800 [#13800] feat(policy): Flat peer SerDe, DTO round-trips, minIntervalMs defaults (#13887) Split from #13801 (2/2). **Stack on #13885** (merge that first). Tracked by #13888. Incremental scope (~1.5k vs #13885 ~1.1k): - Flat JSON mapping via `ExecutablePolicyContentSerde` + Iceberg `PolicyContentDTO` / `DTOConverters` - REST + `POConverters` round-trips for each executable built-in type - `MaintenancePolicyDefinitions` + per-type `gravitino.maintenance.*.minIntervalMs` + conf/docs Until #13885 merges, the GitHub file list includes domain files from the stack base; review the **second commit** / `common`+`core/maintenance`+`conf`+`docs` paths. Preserve flat wire/persistence JSON without flattening Spec into every type; server defaults for omitted `minIntervalMs`. Fix: #13888 Yes — optional JSON peer fields on Iceberg maintenance policy contents, and: - `gravitino.maintenance.compaction.minIntervalMs` (`3600000`) - `gravitino.maintenance.manifest-rewrite.minIntervalMs` (`3600000`) - `gravitino.maintenance.snapshot-expiry.minIntervalMs` (`3600000`) - `gravitino.maintenance.orphan-cleanup.minIntervalMs` (`86400000`) ```bash ./gradlew :common:test --tests 'org.apache.gravitino.dto.policy.*' \ :core:test --tests 'org.apache.gravitino.storage.relational.utils.TestPOConverters' \ --tests 'org.apache.gravitino.maintenance.policy.TestMaintenanceMinIntervalConfigs' \ -PskipITs ``` --------- Co-authored-by: Cursor <[email protected]> --- .../gravitino/dto/policy/PolicyContentDTO.java | 310 ++++++++++++++++++++- .../apache/gravitino/dto/util/DTOConverters.java | 39 ++- .../dto/policy/TestOrphanFileRemovalPolicyDTO.java | 29 ++ .../apache/gravitino/dto/policy/TestPolicyDTO.java | 83 ++++++ ...iteManifestsAndSnapshotExpirationPolicyDTO.java | 45 +++ conf/gravitino.conf.template | 8 + .../policy/MaintenanceMinIntervalConfigs.java | 142 ++++++++++ .../policy/MaintenancePolicyDefinition.java | 62 +++++ .../policy/MaintenancePolicyDefinitions.java | 100 +++++++ .../policy/TestMaintenanceMinIntervalConfigs.java | 134 +++++++++ docs/gravitino-server-config.md | 12 + 11 files changed, 929 insertions(+), 35 deletions(-) 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 323d1d3efa..990c51cf29 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 @@ -18,12 +18,15 @@ */ package org.apache.gravitino.dto.policy; +import com.fasterxml.jackson.annotation.JsonIgnore; +import com.fasterxml.jackson.annotation.JsonInclude; import com.fasterxml.jackson.annotation.JsonProperty; import com.fasterxml.jackson.core.JsonParser; import com.fasterxml.jackson.core.JsonToken; import com.fasterxml.jackson.databind.DeserializationContext; import com.fasterxml.jackson.databind.JsonDeserializer; import com.fasterxml.jackson.databind.annotation.JsonDeserialize; +import com.google.common.base.Preconditions; import java.io.IOException; import java.util.ArrayList; import java.util.Collections; @@ -46,11 +49,100 @@ import org.apache.gravitino.policy.IcebergRewriteManifestsContent; import org.apache.gravitino.policy.IcebergSnapshotExpirationContent; import org.apache.gravitino.policy.PolicyContent; import org.apache.gravitino.policy.PolicyContents; +import org.apache.gravitino.policy.PolicyExecutionSpec; import org.apache.gravitino.policy.RowFilterContent; /** Represents a Policy Content Data Transfer Object (DTO). */ public interface PolicyContentDTO extends PolicyContent { + /** + * Optional maintenance execution peers on Iceberg TMS policy content DTOs. + * + * <p>Peers are stored as a composed {@link PolicyExecutionSpec}. Flat JSON keys are exposed via + * {@code @JsonProperty} accessors so the wire contract stays flat without duplicating five fields + * on every DTO. + * + * <p>Option maps always return unmodifiable defensive copies so callers cannot mutate DTO state + * after validation or change Lombok equality/hash state through the returned map. + */ + interface ExecutionPeersDTO { + + /** + * Stores {@code null} when every peer is unset so Lombok equality matches Jackson round-trips. + * + * @param spec peers to store; empty specs are normalized to {@code null} + * @return {@code null} when unset, otherwise the spec + */ + static PolicyExecutionSpec normalize(@Nullable PolicyExecutionSpec spec) { + return spec == null || !spec.hasAny() ? null : spec; + } + + /** + * @return immutable execution peers; never {@code null} + */ + @JsonIgnore + PolicyExecutionSpec executionSpec(); + + /** + * Optional cron expression for automated evaluation. + * + * @return cron string, or {@code null} when unset + */ + @Nullable + @JsonProperty("timeTrigger") + @JsonInclude(JsonInclude.Include.NON_NULL) + default String timeTrigger() { + return executionSpec().timeTrigger(); + } + + /** + * Optional flag to evaluate the policy after an Iceberg commit. + * + * @return {@code true} when enabled, {@code false} when explicitly disabled, or {@code null} + * when unset + */ + @Nullable + @JsonProperty("commitTrigger") + @JsonInclude(JsonInclude.Include.NON_NULL) + default Boolean commitTrigger() { + return executionSpec().commitTrigger(); + } + + /** + * Optional cooldown between finished automated runs. + * + * @return milliseconds, or {@code null} when unset + */ + @Nullable + @JsonProperty("minIntervalMs") + @JsonInclude(JsonInclude.Include.NON_NULL) + default Long minIntervalMs() { + return executionSpec().minIntervalMs(); + } + + /** + * Optional non-secret job / client options. + * + * @return unmodifiable defensive copy; never {@code null} (empty when unset) + */ + @JsonProperty("jobOptions") + @JsonInclude(JsonInclude.Include.NON_EMPTY) + default Map<String, String> jobOptions() { + return executionSpec().jobOptions(); + } + + /** + * Optional non-secret updater / control-plane options. + * + * @return unmodifiable defensive copy; never {@code null} (empty when unset) + */ + @JsonProperty("updaterOptions") + @JsonInclude(JsonInclude.Include.NON_EMPTY) + default Map<String, String> updaterOptions() { + return executionSpec().updaterOptions(); + } + } + /** Represents a custom policy content DTO. */ @EqualsAndHashCode @ToString @@ -97,10 +189,10 @@ public interface PolicyContentDTO extends PolicyContent { /** Represents a typed iceberg compaction policy content DTO. */ @EqualsAndHashCode - @ToString + @ToString(exclude = {"executionSpec", "rewriteOptions"}) @Builder(setterPrefix = "with") @AllArgsConstructor(access = lombok.AccessLevel.PRIVATE) - class IcebergCompactionContentDTO implements PolicyContentDTO { + class IcebergCompactionContentDTO implements PolicyContentDTO, ExecutionPeersDTO { @JsonProperty("minDataFileMse") private Long minDataFileMse; @@ -126,6 +218,13 @@ public interface PolicyContentDTO extends PolicyContent { @JsonProperty("rewriteOptions") private Map<String, String> rewriteOptions; + @EqualsAndHashCode.Exclude @JsonIgnore @Nullable private PolicyExecutionSpec executionSpec; + + @EqualsAndHashCode.Include + private PolicyExecutionSpec executionSpecForEquals() { + return ExecutionPeersDTO.normalize(executionSpec); + } + // Default constructor for Jackson deserialization only. private IcebergCompactionContentDTO() {} @@ -215,6 +314,41 @@ public interface PolicyContentDTO extends PolicyContent { : Collections.unmodifiableMap(new LinkedHashMap<>(rewriteOptions)); } + @Override + @JsonIgnore + public PolicyExecutionSpec executionSpec() { + return executionSpec == null ? PolicyExecutionSpec.empty() : executionSpec; + } + + @JsonProperty("timeTrigger") + private void setTimeTrigger(String timeTrigger) { + this.executionSpec = + ExecutionPeersDTO.normalize(executionSpec().withTimeTrigger(timeTrigger)); + } + + @JsonProperty("commitTrigger") + private void setCommitTrigger(Boolean commitTrigger) { + this.executionSpec = + ExecutionPeersDTO.normalize(executionSpec().withCommitTrigger(commitTrigger)); + } + + @JsonProperty("minIntervalMs") + private void setMinIntervalMs(Long minIntervalMs) { + this.executionSpec = + ExecutionPeersDTO.normalize(executionSpec().withMinIntervalMs(minIntervalMs)); + } + + @JsonProperty("jobOptions") + private void setJobOptions(Map<String, String> jobOptions) { + this.executionSpec = ExecutionPeersDTO.normalize(executionSpec().withJobOptions(jobOptions)); + } + + @JsonProperty("updaterOptions") + private void setUpdaterOptions(Map<String, String> updaterOptions) { + this.executionSpec = + ExecutionPeersDTO.normalize(executionSpec().withUpdaterOptions(updaterOptions)); + } + @Override public Set<MetadataObject.Type> supportedObjectTypes() { return toDomainContent().supportedObjectTypes(); @@ -236,7 +370,11 @@ public interface PolicyContentDTO extends PolicyContent { toDomainContent().validate(); } - private PolicyContent toDomainContent() { + /** + * @return domain policy content including execution peers + */ + public PolicyContent toDomainContent() { + PolicyExecutionSpec spec = executionSpec(); return PolicyContents.icebergDataCompaction( minDataFileMse(), minDeleteFileNumber(), @@ -245,16 +383,21 @@ public interface PolicyContentDTO extends PolicyContent { maxPartitionNum(), rewriteStrategy(), sortOrder(), - rewriteOptions()); + rewriteOptions(), + spec.timeTrigger(), + spec.commitTrigger(), + spec.minIntervalMs(), + spec.jobOptions(), + spec.updaterOptions()); } } /** Typed orphan cleanup policy content for REST requests and responses. */ @EqualsAndHashCode - @ToString + @ToString(exclude = {"executionSpec"}) @Builder(setterPrefix = "with") @AllArgsConstructor(access = lombok.AccessLevel.PRIVATE) - class IcebergOrphanFileRemovalContentDTO implements PolicyContentDTO { + class IcebergOrphanFileRemovalContentDTO implements PolicyContentDTO, ExecutionPeersDTO { @JsonProperty("olderThanDays") private Long olderThanDays; @@ -264,6 +407,13 @@ public interface PolicyContentDTO extends PolicyContent { @JsonProperty("dryRun") private Boolean dryRun; + @EqualsAndHashCode.Exclude @JsonIgnore @Nullable private PolicyExecutionSpec executionSpec; + + @EqualsAndHashCode.Include + private PolicyExecutionSpec executionSpecForEquals() { + return ExecutionPeersDTO.normalize(executionSpec); + } + private IcebergOrphanFileRemovalContentDTO() {} /** @@ -290,6 +440,41 @@ public interface PolicyContentDTO extends PolicyContent { return dryRun == null ? IcebergOrphanFileRemovalContent.DEFAULT_DRY_RUN : dryRun; } + @Override + @JsonIgnore + public PolicyExecutionSpec executionSpec() { + return executionSpec == null ? PolicyExecutionSpec.empty() : executionSpec; + } + + @JsonProperty("timeTrigger") + private void setTimeTrigger(String timeTrigger) { + this.executionSpec = + ExecutionPeersDTO.normalize(executionSpec().withTimeTrigger(timeTrigger)); + } + + @JsonProperty("commitTrigger") + private void setCommitTrigger(Boolean commitTrigger) { + this.executionSpec = + ExecutionPeersDTO.normalize(executionSpec().withCommitTrigger(commitTrigger)); + } + + @JsonProperty("minIntervalMs") + private void setMinIntervalMs(Long minIntervalMs) { + this.executionSpec = + ExecutionPeersDTO.normalize(executionSpec().withMinIntervalMs(minIntervalMs)); + } + + @JsonProperty("jobOptions") + private void setJobOptions(Map<String, String> jobOptions) { + this.executionSpec = ExecutionPeersDTO.normalize(executionSpec().withJobOptions(jobOptions)); + } + + @JsonProperty("updaterOptions") + private void setUpdaterOptions(Map<String, String> updaterOptions) { + this.executionSpec = + ExecutionPeersDTO.normalize(executionSpec().withUpdaterOptions(updaterOptions)); + } + @Override public Set<MetadataObject.Type> supportedObjectTypes() { return toDomainContent().supportedObjectTypes(); @@ -307,21 +492,36 @@ public interface PolicyContentDTO extends PolicyContent { @Override public void validate() { + Preconditions.checkArgument( + !Boolean.TRUE.equals(commitTrigger()), + "commitTrigger is not allowed for orphan file removal"); PolicyContentDTO.super.validate(); toDomainContent().validate(); } - private PolicyContent toDomainContent() { - return PolicyContents.icebergOrphanFileRemoval(olderThanDays(), location(), dryRun()); + /** + * @return domain policy content including execution peers + */ + public PolicyContent toDomainContent() { + PolicyExecutionSpec spec = executionSpec(); + return PolicyContents.icebergOrphanFileRemoval( + olderThanDays(), + location(), + dryRun(), + spec.timeTrigger(), + spec.commitTrigger(), + spec.minIntervalMs(), + spec.jobOptions(), + spec.updaterOptions()); } } /** Typed manifest rewrite policy content. */ @EqualsAndHashCode - @ToString + @ToString(exclude = {"executionSpec"}) @Builder(setterPrefix = "with") @AllArgsConstructor(access = AccessLevel.PRIVATE) - class IcebergRewriteManifestsContentDTO implements PolicyContentDTO { + class IcebergRewriteManifestsContentDTO implements PolicyContentDTO, ExecutionPeersDTO { @JsonProperty("manifest_count_critical") @JsonDeserialize(using = ManifestThresholdDeserializer.class) private Long manifestCountCritical; @@ -342,6 +542,13 @@ public interface PolicyContentDTO extends PolicyContent { @JsonDeserialize(using = ManifestUseCachingDeserializer.class) private Boolean useCaching; + @EqualsAndHashCode.Exclude @JsonIgnore @Nullable private PolicyExecutionSpec executionSpec; + + @EqualsAndHashCode.Include + private PolicyExecutionSpec executionSpecForEquals() { + return ExecutionPeersDTO.normalize(executionSpec); + } + private IcebergRewriteManifestsContentDTO() {} /** @@ -406,6 +613,41 @@ public interface PolicyContentDTO extends PolicyContent { return useCaching; } + @Override + @JsonIgnore + public PolicyExecutionSpec executionSpec() { + return executionSpec == null ? PolicyExecutionSpec.empty() : executionSpec; + } + + @JsonProperty("timeTrigger") + private void setTimeTrigger(String timeTrigger) { + this.executionSpec = + ExecutionPeersDTO.normalize(executionSpec().withTimeTrigger(timeTrigger)); + } + + @JsonProperty("commitTrigger") + private void setCommitTrigger(Boolean commitTrigger) { + this.executionSpec = + ExecutionPeersDTO.normalize(executionSpec().withCommitTrigger(commitTrigger)); + } + + @JsonProperty("minIntervalMs") + private void setMinIntervalMs(Long minIntervalMs) { + this.executionSpec = + ExecutionPeersDTO.normalize(executionSpec().withMinIntervalMs(minIntervalMs)); + } + + @JsonProperty("jobOptions") + private void setJobOptions(Map<String, String> jobOptions) { + this.executionSpec = ExecutionPeersDTO.normalize(executionSpec().withJobOptions(jobOptions)); + } + + @JsonProperty("updaterOptions") + private void setUpdaterOptions(Map<String, String> updaterOptions) { + this.executionSpec = + ExecutionPeersDTO.normalize(executionSpec().withUpdaterOptions(updaterOptions)); + } + @Override public Set<MetadataObject.Type> supportedObjectTypes() { return toDomainContent().supportedObjectTypes(); @@ -464,13 +706,22 @@ public interface PolicyContentDTO extends PolicyContent { } } - private PolicyContent toDomainContent() { + /** + * @return domain policy content including thresholds and execution peers + */ + public PolicyContent toDomainContent() { + PolicyExecutionSpec spec = executionSpec(); return PolicyContents.icebergRewriteManifests( manifestCountCritical(), manifestCountWarning(), avgManifestSizeThresholdBytes(), specId(), - useCaching()); + useCaching(), + spec.timeTrigger(), + spec.commitTrigger(), + spec.minIntervalMs(), + spec.jobOptions(), + spec.updaterOptions()); } } @@ -524,6 +775,41 @@ public interface PolicyContentDTO extends PolicyContent { : streamResults; } + @Override + @JsonIgnore + public PolicyExecutionSpec executionSpec() { + return executionSpec == null ? PolicyExecutionSpec.empty() : executionSpec; + } + + @JsonProperty("timeTrigger") + private void setTimeTrigger(String timeTrigger) { + this.executionSpec = + ExecutionPeersDTO.normalize(executionSpec().withTimeTrigger(timeTrigger)); + } + + @JsonProperty("commitTrigger") + private void setCommitTrigger(Boolean commitTrigger) { + this.executionSpec = + ExecutionPeersDTO.normalize(executionSpec().withCommitTrigger(commitTrigger)); + } + + @JsonProperty("minIntervalMs") + private void setMinIntervalMs(Long minIntervalMs) { + this.executionSpec = + ExecutionPeersDTO.normalize(executionSpec().withMinIntervalMs(minIntervalMs)); + } + + @JsonProperty("jobOptions") + private void setJobOptions(Map<String, String> jobOptions) { + this.executionSpec = ExecutionPeersDTO.normalize(executionSpec().withJobOptions(jobOptions)); + } + + @JsonProperty("updaterOptions") + private void setUpdaterOptions(Map<String, String> updaterOptions) { + this.executionSpec = + ExecutionPeersDTO.normalize(executionSpec().withUpdaterOptions(updaterOptions)); + } + @Override public Set<MetadataObject.Type> supportedObjectTypes() { return toDomainContent().supportedObjectTypes(); 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 e6996d9f66..19e0072c87 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 @@ -106,12 +106,14 @@ import org.apache.gravitino.model.Model; import org.apache.gravitino.model.ModelVersion; import org.apache.gravitino.policy.AccessControlContent; import org.apache.gravitino.policy.ColumnMaskContent; +import org.apache.gravitino.policy.ExecutablePolicyContent; import org.apache.gravitino.policy.IcebergDataCompactionContent; import org.apache.gravitino.policy.IcebergOrphanFileRemovalContent; import org.apache.gravitino.policy.IcebergRewriteManifestsContent; import org.apache.gravitino.policy.IcebergSnapshotExpirationContent; import org.apache.gravitino.policy.PolicyContent; import org.apache.gravitino.policy.PolicyContents; +import org.apache.gravitino.policy.PolicyExecutionSpec; import org.apache.gravitino.policy.RowFilterContent; import org.apache.gravitino.rel.Column; import org.apache.gravitino.rel.Representation; @@ -672,6 +674,12 @@ public class DTOConverters { return (PolicyContentDTO) policyContent; } + PolicyExecutionSpec executionSpec = null; + if (policyContent instanceof ExecutablePolicyContent) { + PolicyExecutionSpec spec = ((ExecutablePolicyContent) policyContent).executionSpec(); + executionSpec = spec.hasAny() ? spec : null; + } + if (policyContent instanceof PolicyContents.CustomContent) { PolicyContents.CustomContent customContent = (PolicyContents.CustomContent) policyContent; return PolicyContentDTO.CustomContentDTO.builder() @@ -687,6 +695,7 @@ public class DTOConverters { .withOlderThanDays(content.olderThanDays()) .withLocation(content.location()) .withDryRun(content.dryRun()) + .withExecutionSpec(executionSpec) .build(); } if (policyContent instanceof IcebergRewriteManifestsContent) { @@ -697,6 +706,7 @@ public class DTOConverters { .withAvgManifestSizeThresholdBytes(content.avgManifestSizeThresholdBytes()) .withSpecId(content.specId()) .withUseCaching(content.useCaching()) + .withExecutionSpec(executionSpec) .build(); } if (policyContent instanceof IcebergSnapshotExpirationContent) { @@ -719,6 +729,7 @@ public class DTOConverters { .withRewriteStrategy(icebergCompactionContent.rewriteStrategy()) .withSortOrder(icebergCompactionContent.sortOrder()) .withRewriteOptions(icebergCompactionContent.rewriteOptions()) + .withExecutionSpec(executionSpec) .build(); } @@ -1589,20 +1600,12 @@ public class DTOConverters { } if (policyContentDTO instanceof PolicyContentDTO.IcebergOrphanFileRemovalContentDTO) { - PolicyContentDTO.IcebergOrphanFileRemovalContentDTO content = - (PolicyContentDTO.IcebergOrphanFileRemovalContentDTO) policyContentDTO; - return PolicyContents.icebergOrphanFileRemoval( - content.olderThanDays(), content.location(), content.dryRun()); + return ((PolicyContentDTO.IcebergOrphanFileRemovalContentDTO) policyContentDTO) + .toDomainContent(); } if (policyContentDTO instanceof PolicyContentDTO.IcebergRewriteManifestsContentDTO) { - PolicyContentDTO.IcebergRewriteManifestsContentDTO content = - (PolicyContentDTO.IcebergRewriteManifestsContentDTO) policyContentDTO; - return PolicyContents.icebergRewriteManifests( - content.manifestCountCritical(), - content.manifestCountWarning(), - content.avgManifestSizeThresholdBytes(), - content.specId(), - content.useCaching()); + return ((PolicyContentDTO.IcebergRewriteManifestsContentDTO) policyContentDTO) + .toDomainContent(); } if (policyContentDTO instanceof PolicyContentDTO.IcebergSnapshotExpirationContentDTO) { PolicyContentDTO.IcebergSnapshotExpirationContentDTO content = @@ -1611,17 +1614,7 @@ public class DTOConverters { content.olderThanDays(), content.retainLast(), content.streamResults()); } if (policyContentDTO instanceof PolicyContentDTO.IcebergCompactionContentDTO) { - PolicyContentDTO.IcebergCompactionContentDTO icebergCompactionContentDTO = - (PolicyContentDTO.IcebergCompactionContentDTO) policyContentDTO; - return PolicyContents.icebergDataCompaction( - icebergCompactionContentDTO.minDataFileMse(), - icebergCompactionContentDTO.minDeleteFileNumber(), - icebergCompactionContentDTO.dataFileMseWeight(), - icebergCompactionContentDTO.deleteFileNumberWeight(), - icebergCompactionContentDTO.maxPartitionNum(), - icebergCompactionContentDTO.rewriteStrategy(), - icebergCompactionContentDTO.sortOrder(), - icebergCompactionContentDTO.rewriteOptions()); + return ((PolicyContentDTO.IcebergCompactionContentDTO) policyContentDTO).toDomainContent(); } if (policyContentDTO instanceof PolicyContentDTO.RowFilterContentDTO) { diff --git a/common/src/test/java/org/apache/gravitino/dto/policy/TestOrphanFileRemovalPolicyDTO.java b/common/src/test/java/org/apache/gravitino/dto/policy/TestOrphanFileRemovalPolicyDTO.java index 567270c553..4286f7f31c 100644 --- a/common/src/test/java/org/apache/gravitino/dto/policy/TestOrphanFileRemovalPolicyDTO.java +++ b/common/src/test/java/org/apache/gravitino/dto/policy/TestOrphanFileRemovalPolicyDTO.java @@ -18,6 +18,7 @@ */ package org.apache.gravitino.dto.policy; +import com.google.common.collect.ImmutableMap; import java.time.Instant; import org.apache.gravitino.dto.AuditDTO; import org.apache.gravitino.dto.requests.PolicyCreateRequest; @@ -85,6 +86,34 @@ class TestOrphanFileRemovalPolicyDTO { Assertions.assertEquals(domain, DTOConverters.fromDTO(restored.content())); } + @Test + void responseRoundTripWithExecutionPeers() throws Exception { + PolicyContent domain = + 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")); + PolicyDTO dto = + PolicyDTO.builder() + .withName("cleanup") + .withPolicyType("system_iceberg_orphan_file_removal") + .withEnabled(true) + .withContent(DTOConverters.toDTO(domain)) + .withAudit(AuditDTO.builder().withCreator("test").withCreateTime(Instant.EPOCH).build()) + .build(); + String json = JsonUtils.objectMapper().writeValueAsString(dto); + Assertions.assertTrue(json.contains("\"timeTrigger\"")); + Assertions.assertFalse(json.contains("\"executionSpec\"")); + PolicyDTO restored = JsonUtils.objectMapper().readValue(json, PolicyDTO.class); + Assertions.assertEquals(dto, restored); + Assertions.assertEquals(domain, DTOConverters.fromDTO(restored.content())); + } + @Test void createAndUpdateValidateRetentionUpperBoundary() throws Exception { long maximum = IcebergOrphanFileRemovalContent.MAX_OLDER_THAN_DAYS; 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 1f8530f3e0..666f46975b 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 @@ -31,6 +31,7 @@ import org.apache.gravitino.json.JsonUtils; import org.apache.gravitino.policy.IcebergDataCompactionContent; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +import org.apache.gravitino.policy.PolicyExecutionSpec; public class TestPolicyDTO { @@ -239,4 +240,86 @@ public class TestPolicyDTO { Assertions.assertEquals(ImmutableList.of("analyst"), contentDTO.applicableRoles()); Assertions.assertDoesNotThrow(contentDTO::validate); } + + @Test + public void testIcebergCompactionMaintenancePeerFieldsSerDe() throws JsonProcessingException { + AuditDTO audit = AuditDTO.builder().withCreator("user1").withCreateTime(Instant.now()).build(); + PolicyContentDTO.IcebergCompactionContentDTO typedContent = + PolicyContentDTO.IcebergCompactionContentDTO.builder() + .withExecutionSpec( + PolicyExecutionSpec.of( + "0 0 2 * * ?", + true, + 3_600_000L, + ImmutableMap.of("spark.executor.memory", "4g"), + ImmutableMap.of("metalake", "ml"))) + .build(); + + PolicyDTO policyDTO = + PolicyDTO.builder() + .withName("iceberg-compaction-tms") + .withComment("tms peers") + .withPolicyType("system_iceberg_compaction") + .withEnabled(true) + .withContent(typedContent) + .withAudit(audit) + .build(); + + String serJson = JsonUtils.objectMapper().writeValueAsString(policyDTO); + Assertions.assertTrue(serJson.contains("\"timeTrigger\"")); + Assertions.assertFalse(serJson.contains("\"executionSpec\"")); + PolicyDTO deserPolicyDTO = JsonUtils.objectMapper().readValue(serJson, PolicyDTO.class); + Assertions.assertEquals(policyDTO, deserPolicyDTO); + + PolicyContentDTO.IcebergCompactionContentDTO content = + (PolicyContentDTO.IcebergCompactionContentDTO) deserPolicyDTO.content(); + Assertions.assertEquals("0 0 2 * * ?", content.timeTrigger()); + Assertions.assertEquals(Boolean.TRUE, content.commitTrigger()); + Assertions.assertEquals(3_600_000L, content.minIntervalMs()); + Assertions.assertEquals("4g", content.jobOptions().get("spark.executor.memory")); + Assertions.assertEquals("ml", content.updaterOptions().get("metalake")); + Assertions.assertDoesNotThrow(content::validate); + } + + @Test + public void testExecutionPeerOptionMapsAreUnmodifiableCopies() { + PolicyContentDTO.IcebergCompactionContentDTO content = + PolicyContentDTO.IcebergCompactionContentDTO.builder() + .withExecutionSpec( + PolicyExecutionSpec.of( + null, + null, + null, + ImmutableMap.of("spark.executor.memory", "4g"), + ImmutableMap.of("metalake", "ml"))) + .build(); + + Assertions.assertThrows( + UnsupportedOperationException.class, + () -> content.jobOptions().put("spark.executor.memory", "8g")); + Assertions.assertThrows( + UnsupportedOperationException.class, + () -> content.updaterOptions().put("metalake", "other")); + Assertions.assertEquals("4g", content.jobOptions().get("spark.executor.memory")); + Assertions.assertEquals("ml", content.updaterOptions().get("metalake")); + } + + @Test + public void testMaintenancePeerValidateOnDto() { + PolicyContentDTO.IcebergOrphanFileRemovalContentDTO orphanOnCommit = + PolicyContentDTO.IcebergOrphanFileRemovalContentDTO.builder() + .withExecutionSpec(PolicyExecutionSpec.of(null, true, null, null, null)) + .build(); + IllegalArgumentException orphanEx = + Assertions.assertThrows(IllegalArgumentException.class, orphanOnCommit::validate); + Assertions.assertTrue(orphanEx.getMessage().contains("commitTrigger")); + + // Cron syntax is not validated on the policy API; scheduler registration parses it. + PolicyContentDTO.IcebergCompactionContentDTO opaqueCron = + PolicyContentDTO.IcebergCompactionContentDTO.builder() + .withExecutionSpec(PolicyExecutionSpec.of("a a a a a a", null, null, null, null)) + .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 2087eef48a..3ce7b9ea05 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 @@ -18,6 +18,9 @@ */ package org.apache.gravitino.dto.policy; +import com.google.common.collect.ImmutableMap; +import java.time.Instant; +import org.apache.gravitino.dto.AuditDTO; import org.apache.gravitino.dto.requests.PolicyCreateRequest; import org.apache.gravitino.dto.requests.PolicyUpdateRequest; import org.apache.gravitino.dto.util.DTOConverters; @@ -41,6 +44,30 @@ class TestRewriteManifestsAndSnapshotExpirationPolicyDTO { "system_iceberg_snapshot_expiration", PolicyContents.icebergSnapshotExpiration()); } + @Test + void rewriteManifestsResponseRoundTripWithExecutionPeers() throws Exception { + assertResponseRoundTripWithPeers( + "system_iceberg_rewrite_manifests", + PolicyContents.icebergRewriteManifests( + "0 30 1 * * ?", + true, + 900_000L, + ImmutableMap.of("spark.executor.memory", "2g"), + ImmutableMap.of("queue", "tms"))); + } + + @Test + void snapshotExpirationResponseRoundTripWithExecutionPeers() throws Exception { + assertResponseRoundTripWithPeers( + "system_iceberg_snapshot_expiration", + PolicyContents.icebergSnapshotExpiration( + "0 0 4 * * ?", + false, + 2_700_000L, + ImmutableMap.of("spark.driver.memory", "1g"), + ImmutableMap.of("metalake", "ml"))); + } + private static void assertCreateAndUpdateRoundTrip(String policyType, PolicyContent expected) throws Exception { PolicyCreateRequest create = @@ -65,4 +92,22 @@ class TestRewriteManifestsAndSnapshotExpirationPolicyDTO { (PolicyUpdateRequest.UpdatePolicyContentRequest) update; Assertions.assertEquals(expected, DTOConverters.fromDTO(contentUpdate.getNewContent())); } + + private static void assertResponseRoundTripWithPeers(String policyType, PolicyContent domain) + throws Exception { + PolicyDTO dto = + PolicyDTO.builder() + .withName("p") + .withPolicyType(policyType) + .withEnabled(true) + .withContent(DTOConverters.toDTO(domain)) + .withAudit(AuditDTO.builder().withCreator("test").withCreateTime(Instant.EPOCH).build()) + .build(); + String json = JsonUtils.objectMapper().writeValueAsString(dto); + Assertions.assertTrue(json.contains("\"timeTrigger\"")); + Assertions.assertFalse(json.contains("\"executionSpec\"")); + PolicyDTO restored = JsonUtils.objectMapper().readValue(json, PolicyDTO.class); + Assertions.assertEquals(dto, restored); + Assertions.assertEquals(domain, DTOConverters.fromDTO(restored.content())); + } } diff --git a/conf/gravitino.conf.template b/conf/gravitino.conf.template index ef26e8a0b5..66554eb606 100644 --- a/conf/gravitino.conf.template +++ b/conf/gravitino.conf.template @@ -146,6 +146,14 @@ gravitino.lance-rest.namespace-backend = gravitino # The metalake name used for Lance REST service gravitino namespace backend, please create the metalake first before using it, and configure the metalake name here. # gravitino.lance-rest.gravitino-metalake = metalake +# THE CONFIGURATION FOR table maintenance (TMS) cooldowns +# Used when a maintenance policy omits minIntervalMs. Per-type defaults: +# compaction / manifest-rewrite / snapshot-expiry = 3600000 (1h), orphan-cleanup = 86400000 (1d) +# gravitino.maintenance.compaction.minIntervalMs = 3600000 +# gravitino.maintenance.manifest-rewrite.minIntervalMs = 3600000 +# gravitino.maintenance.snapshot-expiry.minIntervalMs = 3600000 +# gravitino.maintenance.orphan-cleanup.minIntervalMs = 86400000 + # THE CONFIGURATION FOR sensitive property key matching # Comma-separated keywords. Each entry is a literal substring, not a regular expression. # This list replaces the default (secret,password,token,credential,access,account). diff --git a/core/src/main/java/org/apache/gravitino/maintenance/policy/MaintenanceMinIntervalConfigs.java b/core/src/main/java/org/apache/gravitino/maintenance/policy/MaintenanceMinIntervalConfigs.java new file mode 100644 index 0000000000..94dd81bae7 --- /dev/null +++ b/core/src/main/java/org/apache/gravitino/maintenance/policy/MaintenanceMinIntervalConfigs.java @@ -0,0 +1,142 @@ +/* + * 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.policy; + +import com.google.common.base.Preconditions; +import javax.annotation.Nullable; +import org.apache.gravitino.Config; +import org.apache.gravitino.config.ConfigBuilder; +import org.apache.gravitino.config.ConfigConstants; +import org.apache.gravitino.config.ConfigEntry; +import org.apache.gravitino.policy.Policy; + +/** + * Server config entries for per-type {@code minIntervalMs} cooldowns in {@code gravitino.conf}. + * + * <p>Used when a maintenance policy omits {@code minIntervalMs}. Resolution order is policy content + * → these server keys → the code defaults below. + * + * <p>Lookup by {@link Policy.BuiltInType} goes through {@link MaintenancePolicyDefinitions}; do not + * add a parallel switch or task-type enum for the same 1:1 mapping. + */ +public final class MaintenanceMinIntervalConfigs { + + /** Default cooldown for compaction / manifest-rewrite / snapshot-expiry (1 hour). */ + public static final long DEFAULT_HOURLY_MIN_INTERVAL_MS = 3_600_000L; + + /** Default cooldown for orphan-cleanup (1 day). */ + public static final long DEFAULT_ORPHAN_CLEANUP_MIN_INTERVAL_MS = 86_400_000L; + + /** Config key for compaction cooldown. */ + public static final String COMPACTION_MIN_INTERVAL_MS_KEY = + "gravitino.maintenance.compaction.minIntervalMs"; + + /** Config key for manifest-rewrite cooldown. */ + public static final String MANIFEST_REWRITE_MIN_INTERVAL_MS_KEY = + "gravitino.maintenance.manifest-rewrite.minIntervalMs"; + + /** Config key for snapshot-expiry cooldown. */ + public static final String SNAPSHOT_EXPIRY_MIN_INTERVAL_MS_KEY = + "gravitino.maintenance.snapshot-expiry.minIntervalMs"; + + /** Config key for orphan-cleanup cooldown. */ + public static final String ORPHAN_CLEANUP_MIN_INTERVAL_MS_KEY = + "gravitino.maintenance.orphan-cleanup.minIntervalMs"; + + /** Cooldown between finished compaction runs when the policy omits {@code minIntervalMs}. */ + public static final ConfigEntry<Long> COMPACTION_MIN_INTERVAL_MS = + new ConfigBuilder(COMPACTION_MIN_INTERVAL_MS_KEY) + .doc( + "Minimum interval in milliseconds between finished compaction runs when the " + + "policy omits minIntervalMs") + .version(ConfigConstants.VERSION_2_0_0) + .longConf() + .checkValue(v -> v != null && v >= 0, "minIntervalMs must be >= 0") + .createWithDefault(DEFAULT_HOURLY_MIN_INTERVAL_MS); + + /** + * Cooldown between finished manifest-rewrite runs when the policy omits {@code minIntervalMs}. + */ + public static final ConfigEntry<Long> MANIFEST_REWRITE_MIN_INTERVAL_MS = + new ConfigBuilder(MANIFEST_REWRITE_MIN_INTERVAL_MS_KEY) + .doc( + "Minimum interval in milliseconds between finished manifest-rewrite runs when the " + + "policy omits minIntervalMs") + .version(ConfigConstants.VERSION_2_0_0) + .longConf() + .checkValue(v -> v != null && v >= 0, "minIntervalMs must be >= 0") + .createWithDefault(DEFAULT_HOURLY_MIN_INTERVAL_MS); + + /** Cooldown between finished snapshot-expiry runs when the policy omits {@code minIntervalMs}. */ + public static final ConfigEntry<Long> SNAPSHOT_EXPIRY_MIN_INTERVAL_MS = + new ConfigBuilder(SNAPSHOT_EXPIRY_MIN_INTERVAL_MS_KEY) + .doc( + "Minimum interval in milliseconds between finished snapshot-expiry runs when the " + + "policy omits minIntervalMs") + .version(ConfigConstants.VERSION_2_0_0) + .longConf() + .checkValue(v -> v != null && v >= 0, "minIntervalMs must be >= 0") + .createWithDefault(DEFAULT_HOURLY_MIN_INTERVAL_MS); + + /** Cooldown between finished orphan-cleanup runs when the policy omits {@code minIntervalMs}. */ + public static final ConfigEntry<Long> ORPHAN_CLEANUP_MIN_INTERVAL_MS = + new ConfigBuilder(ORPHAN_CLEANUP_MIN_INTERVAL_MS_KEY) + .doc( + "Minimum interval in milliseconds between finished orphan-cleanup runs when the " + + "policy omits minIntervalMs") + .version(ConfigConstants.VERSION_2_0_0) + .longConf() + .checkValue(v -> v != null && v >= 0, "minIntervalMs must be >= 0") + .createWithDefault(DEFAULT_ORPHAN_CLEANUP_MIN_INTERVAL_MS); + + private MaintenanceMinIntervalConfigs() {} + + /** + * Returns the {@code minIntervalMs} config entry for a built-in maintenance policy type. + * + * @param builtInType built-in policy type + * @return config entry + */ + public static ConfigEntry<Long> forBuiltInType(Policy.BuiltInType builtInType) { + return MaintenancePolicyDefinitions.require(builtInType).minIntervalMs(); + } + + /** + * Resolves minimum interval: policy content, then server config entry (with code default). + * + * @param config gravitino server config + * @param builtInType built-in policy type + * @param policyMinIntervalMs value from effective policy content, or null + * @return resolved interval in milliseconds + * @throws IllegalArgumentException if {@code policyMinIntervalMs} is negative + */ + public static long resolve( + Config config, Policy.BuiltInType builtInType, @Nullable Long policyMinIntervalMs) { + Preconditions.checkNotNull(config, "config cannot be null"); + Preconditions.checkNotNull(builtInType, "builtInType cannot be null"); + if (policyMinIntervalMs != null) { + Preconditions.checkArgument( + policyMinIntervalMs >= 0, + "policy minIntervalMs must be >= 0, got %s", + policyMinIntervalMs); + return policyMinIntervalMs; + } + return config.get(forBuiltInType(builtInType)); + } +} diff --git a/core/src/main/java/org/apache/gravitino/maintenance/policy/MaintenancePolicyDefinition.java b/core/src/main/java/org/apache/gravitino/maintenance/policy/MaintenancePolicyDefinition.java new file mode 100644 index 0000000000..e1f2da234c --- /dev/null +++ b/core/src/main/java/org/apache/gravitino/maintenance/policy/MaintenancePolicyDefinition.java @@ -0,0 +1,62 @@ +/* + * 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.policy; + +import com.google.common.base.Preconditions; +import org.apache.gravitino.config.ConfigEntry; +import org.apache.gravitino.policy.Policy; + +/** + * Per-type maintenance metadata keyed by {@link Policy.BuiltInType}. + * + * <p>{@link Policy.BuiltInType} is the single policy/activity identity (not a db-scheduler task + * kind). Server defaults such as {@code minIntervalMs} are looked up from {@link + * MaintenancePolicyDefinitions} rather than parallel switches or a second activity enum. + */ +public final class MaintenancePolicyDefinition { + + private final Policy.BuiltInType builtInType; + private final ConfigEntry<Long> minIntervalMs; + + MaintenancePolicyDefinition(Policy.BuiltInType builtInType, ConfigEntry<Long> minIntervalMs) { + this.builtInType = Preconditions.checkNotNull(builtInType, "builtInType cannot be null"); + this.minIntervalMs = Preconditions.checkNotNull(minIntervalMs, "minIntervalMs cannot be null"); + } + + /** + * @return built-in policy type identity + */ + public Policy.BuiltInType builtInType() { + return builtInType; + } + + /** + * @return server config entry for cooldown when policy content omits {@code minIntervalMs} + */ + public ConfigEntry<Long> minIntervalMs() { + return minIntervalMs; + } + + /** + * @return {@code gravitino.conf} key for this type's cooldown + */ + public String minIntervalMsKey() { + return minIntervalMs.getKey(); + } +} diff --git a/core/src/main/java/org/apache/gravitino/maintenance/policy/MaintenancePolicyDefinitions.java b/core/src/main/java/org/apache/gravitino/maintenance/policy/MaintenancePolicyDefinitions.java new file mode 100644 index 0000000000..bfa3eeae8a --- /dev/null +++ b/core/src/main/java/org/apache/gravitino/maintenance/policy/MaintenancePolicyDefinitions.java @@ -0,0 +1,100 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.gravitino.maintenance.policy; + +import com.google.common.base.Preconditions; +import com.google.common.collect.ImmutableMap; +import java.util.Collection; +import java.util.EnumMap; +import java.util.Map; +import org.apache.gravitino.config.ConfigEntry; +import org.apache.gravitino.policy.Policy; + +/** + * Core registry of Iceberg maintenance policy definitions. + * + * <p>Keyed only by {@link Policy.BuiltInType} (policy / activity identity). This is + * <strong>not</strong> a scheduler task-kind registry — {@code policy-expand} / {@code cron-job} / + * {@code commit-job} stay separate. Non-maintenance built-ins ({@code ROW_FILTER}, {@code + * COLUMN_MASK}, {@code CUSTOM}) are absent from this map. + */ +public final class MaintenancePolicyDefinitions { + + private static final Map<Policy.BuiltInType, MaintenancePolicyDefinition> BY_TYPE; + + static { + EnumMap<Policy.BuiltInType, MaintenancePolicyDefinition> map = + new EnumMap<>(Policy.BuiltInType.class); + register( + map, + Policy.BuiltInType.ICEBERG_COMPACTION, + MaintenanceMinIntervalConfigs.COMPACTION_MIN_INTERVAL_MS); + register( + map, + Policy.BuiltInType.ICEBERG_REWRITE_MANIFESTS, + MaintenanceMinIntervalConfigs.MANIFEST_REWRITE_MIN_INTERVAL_MS); + register( + map, + Policy.BuiltInType.ICEBERG_SNAPSHOT_EXPIRATION, + MaintenanceMinIntervalConfigs.SNAPSHOT_EXPIRY_MIN_INTERVAL_MS); + register( + map, + Policy.BuiltInType.ICEBERG_ORPHAN_FILE_REMOVAL, + MaintenanceMinIntervalConfigs.ORPHAN_CLEANUP_MIN_INTERVAL_MS); + BY_TYPE = ImmutableMap.copyOf(map); + } + + private MaintenancePolicyDefinitions() {} + + /** + * Returns the maintenance definition for a built-in type. + * + * @param builtInType built-in policy type + * @return definition + * @throws IllegalArgumentException if the type is not a registered maintenance policy + */ + public static MaintenancePolicyDefinition require(Policy.BuiltInType builtInType) { + Preconditions.checkNotNull(builtInType, "builtInType cannot be null"); + MaintenancePolicyDefinition definition = BY_TYPE.get(builtInType); + Preconditions.checkArgument( + definition != null, "No maintenance definition for built-in policy type: %s", builtInType); + return definition; + } + + /** + * @return all registered maintenance definitions + */ + public static Collection<MaintenancePolicyDefinition> all() { + return BY_TYPE.values(); + } + + /** + * @return whether {@code builtInType} is a registered Iceberg maintenance policy + */ + public static boolean isMaintenance(Policy.BuiltInType builtInType) { + return builtInType != null && BY_TYPE.containsKey(builtInType); + } + + private static void register( + EnumMap<Policy.BuiltInType, MaintenancePolicyDefinition> map, + Policy.BuiltInType builtInType, + ConfigEntry<Long> minIntervalMs) { + map.put(builtInType, new MaintenancePolicyDefinition(builtInType, minIntervalMs)); + } +} diff --git a/core/src/test/java/org/apache/gravitino/maintenance/policy/TestMaintenanceMinIntervalConfigs.java b/core/src/test/java/org/apache/gravitino/maintenance/policy/TestMaintenanceMinIntervalConfigs.java new file mode 100644 index 0000000000..9caf1ff493 --- /dev/null +++ b/core/src/test/java/org/apache/gravitino/maintenance/policy/TestMaintenanceMinIntervalConfigs.java @@ -0,0 +1,134 @@ +/* + * 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.policy; + +import org.apache.gravitino.Config; +import org.apache.gravitino.policy.Policy; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +public class TestMaintenanceMinIntervalConfigs { + + @Test + void testRegisteredDefaults() { + Config config = new Config(false) {}; + Assertions.assertEquals( + MaintenanceMinIntervalConfigs.DEFAULT_HOURLY_MIN_INTERVAL_MS, + config.get(MaintenanceMinIntervalConfigs.COMPACTION_MIN_INTERVAL_MS)); + Assertions.assertEquals( + MaintenanceMinIntervalConfigs.DEFAULT_HOURLY_MIN_INTERVAL_MS, + config.get(MaintenanceMinIntervalConfigs.MANIFEST_REWRITE_MIN_INTERVAL_MS)); + Assertions.assertEquals( + MaintenanceMinIntervalConfigs.DEFAULT_HOURLY_MIN_INTERVAL_MS, + config.get(MaintenanceMinIntervalConfigs.SNAPSHOT_EXPIRY_MIN_INTERVAL_MS)); + Assertions.assertEquals( + MaintenanceMinIntervalConfigs.DEFAULT_ORPHAN_CLEANUP_MIN_INTERVAL_MS, + config.get(MaintenanceMinIntervalConfigs.ORPHAN_CLEANUP_MIN_INTERVAL_MS)); + } + + @Test + void testConfigKeys() { + Assertions.assertEquals( + MaintenanceMinIntervalConfigs.COMPACTION_MIN_INTERVAL_MS_KEY, + MaintenanceMinIntervalConfigs.forBuiltInType(Policy.BuiltInType.ICEBERG_COMPACTION) + .getKey()); + Assertions.assertEquals( + MaintenanceMinIntervalConfigs.MANIFEST_REWRITE_MIN_INTERVAL_MS_KEY, + MaintenanceMinIntervalConfigs.forBuiltInType(Policy.BuiltInType.ICEBERG_REWRITE_MANIFESTS) + .getKey()); + Assertions.assertEquals( + MaintenanceMinIntervalConfigs.SNAPSHOT_EXPIRY_MIN_INTERVAL_MS_KEY, + MaintenanceMinIntervalConfigs.forBuiltInType(Policy.BuiltInType.ICEBERG_SNAPSHOT_EXPIRATION) + .getKey()); + Assertions.assertEquals( + MaintenanceMinIntervalConfigs.ORPHAN_CLEANUP_MIN_INTERVAL_MS_KEY, + MaintenanceMinIntervalConfigs.forBuiltInType(Policy.BuiltInType.ICEBERG_ORPHAN_FILE_REMOVAL) + .getKey()); + } + + @Test + void testResolvePrefersPolicyThenConfigThenDefault() { + Config config = new Config(false) {}; + config.set(MaintenanceMinIntervalConfigs.COMPACTION_MIN_INTERVAL_MS, 900_000L); + + Assertions.assertEquals( + 120_000L, + MaintenanceMinIntervalConfigs.resolve( + config, Policy.BuiltInType.ICEBERG_COMPACTION, 120_000L)); + Assertions.assertEquals( + 900_000L, + MaintenanceMinIntervalConfigs.resolve(config, Policy.BuiltInType.ICEBERG_COMPACTION, null)); + + Config defaultsOnly = new Config(false) {}; + Assertions.assertEquals( + MaintenanceMinIntervalConfigs.DEFAULT_ORPHAN_CLEANUP_MIN_INTERVAL_MS, + MaintenanceMinIntervalConfigs.resolve( + defaultsOnly, Policy.BuiltInType.ICEBERG_ORPHAN_FILE_REMOVAL, null)); + } + + @Test + void testRejectsNegativeMinInterval() { + Config config = new Config(false) {}; + config.set(MaintenanceMinIntervalConfigs.COMPACTION_MIN_INTERVAL_MS, -1L); + Assertions.assertThrows( + IllegalArgumentException.class, + () -> config.get(MaintenanceMinIntervalConfigs.COMPACTION_MIN_INTERVAL_MS)); + } + + @Test + void testResolveRejectsNegativePolicyMinInterval() { + Config config = new Config(false) {}; + Assertions.assertThrows( + IllegalArgumentException.class, + () -> + MaintenanceMinIntervalConfigs.resolve( + config, Policy.BuiltInType.ICEBERG_COMPACTION, -1L)); + } + + @Test + void testRejectsNonMaintenanceBuiltInType() { + Assertions.assertThrows( + IllegalArgumentException.class, + () -> MaintenanceMinIntervalConfigs.forBuiltInType(Policy.BuiltInType.ROW_FILTER)); + Assertions.assertFalse( + MaintenancePolicyDefinitions.isMaintenance(Policy.BuiltInType.ROW_FILTER)); + } + + @Test + void testRegistryCoversFourMaintenanceTypes() { + Assertions.assertEquals(4, MaintenancePolicyDefinitions.all().size()); + Assertions.assertTrue( + MaintenancePolicyDefinitions.isMaintenance(Policy.BuiltInType.ICEBERG_COMPACTION)); + Assertions.assertTrue( + MaintenancePolicyDefinitions.isMaintenance(Policy.BuiltInType.ICEBERG_REWRITE_MANIFESTS)); + Assertions.assertTrue( + MaintenancePolicyDefinitions.isMaintenance(Policy.BuiltInType.ICEBERG_SNAPSHOT_EXPIRATION)); + Assertions.assertTrue( + MaintenancePolicyDefinitions.isMaintenance(Policy.BuiltInType.ICEBERG_ORPHAN_FILE_REMOVAL)); + + MaintenancePolicyDefinition compaction = + MaintenancePolicyDefinitions.require(Policy.BuiltInType.ICEBERG_COMPACTION); + Assertions.assertEquals(Policy.BuiltInType.ICEBERG_COMPACTION, compaction.builtInType()); + Assertions.assertEquals( + MaintenanceMinIntervalConfigs.COMPACTION_MIN_INTERVAL_MS, compaction.minIntervalMs()); + Assertions.assertEquals( + MaintenanceMinIntervalConfigs.COMPACTION_MIN_INTERVAL_MS_KEY, + compaction.minIntervalMsKey()); + } +} diff --git a/docs/gravitino-server-config.md b/docs/gravitino-server-config.md index a96db19343..ce60086496 100644 --- a/docs/gravitino-server-config.md +++ b/docs/gravitino-server-config.md @@ -659,6 +659,18 @@ server, are documented with those services. See | `gravitino.job.outputMaxLines` | Maximum number of lines returned when fetching a job's stdout/stderr output. | `1000` | | `gravitino.job.outputMaxBytes` | Maximum number of bytes read from the tail of a job's stdout/stderr when fetching its output. | `262144` (256KB) | +#### Table maintenance cooldowns + +When a table-maintenance policy omits `minIntervalMs`, the server uses these per-type keys, then +the code defaults below. Policy content always wins when set. + +| Configuration Item | Description | Default Value | +|--------------------------------------------------------|-------------------------------------------------|-----------------| +| `gravitino.maintenance.compaction.minIntervalMs` | Cooldown between finished compaction runs | `3600000` (1h) | +| `gravitino.maintenance.manifest-rewrite.minIntervalMs` | Cooldown between finished manifest-rewrite runs | `3600000` (1h) | +| `gravitino.maintenance.snapshot-expiry.minIntervalMs` | Cooldown between finished snapshot-expiry runs | `3600000` (1h) | +| `gravitino.maintenance.orphan-cleanup.minIntervalMs` | Cooldown between finished orphan-cleanup runs | `86400000` (1d) | + ### Key Management The server talks to KMS instances you name in `gravitino.conf`. Each name is one configured
