This is an automated email from the ASF dual-hosted git repository.
roryqi pushed a commit to branch branch-tms-scheduler
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/branch-tms-scheduler by this
push:
new 7bbc5b80e0 [#13800] feat(policy): Flat peer SerDe, DTO round-trips,
minIntervalMs defaults (#13887)
7bbc5b80e0 is described below
commit 7bbc5b80e0a372be79a32524d9daa55f2747bc42
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)
### What changes were proposed in this pull request?
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.
### Why are the changes needed?
Preserve flat wire/persistence JSON without flattening Spec into every
type; server defaults for omitted `minIntervalMs`.
Fix: #13888
### Does this PR introduce _any_ user-facing change?
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`)
### How was this patch tested?
```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 | 338 ++++++++++++++++++++-
.../apache/gravitino/dto/util/DTOConverters.java | 42 +--
.../dto/policy/TestOrphanFileRemovalPolicyDTO.java | 29 ++
.../apache/gravitino/dto/policy/TestPolicyDTO.java | 82 +++++
...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, 959 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 4ac19d7be4..7fbd4bc365 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,7 +18,10 @@
*/
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.google.common.base.Preconditions;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.Map;
@@ -34,11 +37,100 @@ import
org.apache.gravitino.policy.IcebergDataCompactionContent;
import org.apache.gravitino.policy.IcebergOrphanFileRemovalContent;
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
@@ -85,10 +177,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;
@@ -114,6 +206,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() {}
@@ -203,6 +302,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();
@@ -224,7 +358,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(),
@@ -233,16 +371,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;
@@ -252,6 +395,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() {}
/**
@@ -278,6 +428,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();
@@ -295,19 +480,43 @@ 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());
}
}
/** Placeholder DTO for Iceberg rewrite-manifests policy content. */
@EqualsAndHashCode
- @ToString
- class IcebergRewriteManifestsContentDTO implements PolicyContentDTO {
+ @ToString(exclude = {"executionSpec"})
+ @Builder(setterPrefix = "with")
+ @AllArgsConstructor(access = lombok.AccessLevel.PRIVATE)
+ class IcebergRewriteManifestsContentDTO implements PolicyContentDTO,
ExecutionPeersDTO {
+ @EqualsAndHashCode.Exclude @JsonIgnore @Nullable private
PolicyExecutionSpec executionSpec;
+
+ @EqualsAndHashCode.Include
+ private PolicyExecutionSpec executionSpecForEquals() {
+ return ExecutionPeersDTO.normalize(executionSpec);
+ }
+
private IcebergRewriteManifestsContentDTO() {}
/**
@@ -317,6 +526,41 @@ public interface PolicyContentDTO extends PolicyContent {
return new IcebergRewriteManifestsContentDTO();
}
+ @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();
@@ -338,15 +582,33 @@ public interface PolicyContentDTO extends PolicyContent {
toDomainContent().validate();
}
- private PolicyContent toDomainContent() {
- return PolicyContents.icebergRewriteManifests();
+ /**
+ * @return domain policy content including execution peers
+ */
+ public PolicyContent toDomainContent() {
+ PolicyExecutionSpec spec = executionSpec();
+ return PolicyContents.icebergRewriteManifests(
+ spec.timeTrigger(),
+ spec.commitTrigger(),
+ spec.minIntervalMs(),
+ spec.jobOptions(),
+ spec.updaterOptions());
}
}
/** Placeholder DTO for Iceberg snapshot-expiration policy content. */
@EqualsAndHashCode
- @ToString
- class IcebergSnapshotExpirationContentDTO implements PolicyContentDTO {
+ @ToString(exclude = {"executionSpec"})
+ @Builder(setterPrefix = "with")
+ @AllArgsConstructor(access = lombok.AccessLevel.PRIVATE)
+ class IcebergSnapshotExpirationContentDTO implements PolicyContentDTO,
ExecutionPeersDTO {
+ @EqualsAndHashCode.Exclude @JsonIgnore @Nullable private
PolicyExecutionSpec executionSpec;
+
+ @EqualsAndHashCode.Include
+ private PolicyExecutionSpec executionSpecForEquals() {
+ return ExecutionPeersDTO.normalize(executionSpec);
+ }
+
private IcebergSnapshotExpirationContentDTO() {}
/**
@@ -356,6 +618,41 @@ public interface PolicyContentDTO extends PolicyContent {
return new IcebergSnapshotExpirationContentDTO();
}
+ @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();
@@ -377,8 +674,17 @@ public interface PolicyContentDTO extends PolicyContent {
toDomainContent().validate();
}
- private PolicyContent toDomainContent() {
- return PolicyContents.icebergSnapshotExpiration();
+ /**
+ * @return domain policy content including execution peers
+ */
+ public PolicyContent toDomainContent() {
+ PolicyExecutionSpec spec = executionSpec();
+ return PolicyContents.icebergSnapshotExpiration(
+ spec.timeTrigger(),
+ spec.commitTrigger(),
+ spec.minIntervalMs(),
+ spec.jobOptions(),
+ spec.updaterOptions());
}
}
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 ebf54be030..019b817fb0 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
@@ -105,12 +105,14 @@ import org.apache.gravitino.messaging.Topic;
import org.apache.gravitino.model.Model;
import org.apache.gravitino.model.ModelVersion;
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;
@@ -671,6 +673,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()
@@ -686,13 +694,18 @@ public class DTOConverters {
.withOlderThanDays(content.olderThanDays())
.withLocation(content.location())
.withDryRun(content.dryRun())
+ .withExecutionSpec(executionSpec)
.build();
}
if (policyContent instanceof IcebergRewriteManifestsContent) {
- return PolicyContentDTO.IcebergRewriteManifestsContentDTO.instance();
+ return PolicyContentDTO.IcebergRewriteManifestsContentDTO.builder()
+ .withExecutionSpec(executionSpec)
+ .build();
}
if (policyContent instanceof IcebergSnapshotExpirationContent) {
- return PolicyContentDTO.IcebergSnapshotExpirationContentDTO.instance();
+ return PolicyContentDTO.IcebergSnapshotExpirationContentDTO.builder()
+ .withExecutionSpec(executionSpec)
+ .build();
}
if (policyContent instanceof IcebergDataCompactionContent) {
IcebergDataCompactionContent icebergCompactionContent =
@@ -706,6 +719,7 @@ public class DTOConverters {
.withRewriteStrategy(icebergCompactionContent.rewriteStrategy())
.withSortOrder(icebergCompactionContent.sortOrder())
.withRewriteOptions(icebergCompactionContent.rewriteOptions())
+ .withExecutionSpec(executionSpec)
.build();
}
@@ -1568,29 +1582,19 @@ 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) {
- return PolicyContents.icebergRewriteManifests();
+ return ((PolicyContentDTO.IcebergRewriteManifestsContentDTO)
policyContentDTO)
+ .toDomainContent();
}
if (policyContentDTO instanceof
PolicyContentDTO.IcebergSnapshotExpirationContentDTO) {
- return PolicyContents.icebergSnapshotExpiration();
+ return ((PolicyContentDTO.IcebergSnapshotExpirationContentDTO)
policyContentDTO)
+ .toDomainContent();
}
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 8dc0292c10..7086df12fa 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
@@ -27,6 +27,7 @@ import org.apache.gravitino.MetadataObject;
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;
@@ -177,4 +178,85 @@ public class TestPolicyDTO {
IcebergDataCompactionContent.DEFAULT_SORT_ORDER,
contentDTO.sortOrder());
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 af8a6c0651..cf3c5d6eae 100644
--- a/conf/gravitino.conf.template
+++ b/conf/gravitino.conf.template
@@ -144,6 +144,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 f4cf1fe400..1ebe6bc6a0 100644
--- a/docs/gravitino-server-config.md
+++ b/docs/gravitino-server-config.md
@@ -577,6 +577,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
