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 d926cddeb8 [#13800] feat(policy): Maintenance execution peers on
Iceberg contents (#13885)
d926cddeb8 is described below
commit d926cddeb81e276c5404f6a662df2f482f3f7b27
Author: MaSai <[email protected]>
AuthorDate: Fri Oct 16 10:01:18 2026 +0800
[#13800] feat(policy): Maintenance execution peers on Iceberg contents
(#13885)
### What changes were proposed in this pull request?
Split from #13801 (1/2). **Domain** execution peers for Epic #13764 /
#13800.
- `PolicyExecutionSpec` / `ExecutablePolicyContent` /
`PolicyExecutionFields`
- Four Iceberg maintenance contents carry peers at construction
- TMS design notes on peers / `BuiltInType` identity
**Follow-up (stack):** #13888 / #13887 — wire/DTO/SerDe +
`minIntervalMs` registry.
### Why are the changes needed?
TMS needs shared trigger / cooldown / option-bag concepts without
freezing them onto every `PolicyContent`.
Fix: #13800
### Does this PR introduce _any_ user-facing change?
API/domain surface for optional peers on Iceberg maintenance contents
(REST wire lands in #13887).
### How was this patch tested?
```bash
./gradlew :api:test --tests 'org.apache.gravitino.policy.*' -PskipITs
```
---------
Co-authored-by: Cursor <[email protected]>
---
.../policy/AbstractExecutablePolicyContent.java | 112 +++++++
.../gravitino/policy/ExecutablePolicyContent.java | 35 ++
.../policy/IcebergDataCompactionContent.java | 78 +++--
.../policy/IcebergOrphanFileRemovalContent.java | 23 +-
.../policy/IcebergRewriteManifestsContent.java | 23 +-
.../policy/IcebergSnapshotExpirationContent.java | 23 +-
.../org/apache/gravitino/policy/PolicyContent.java | 11 +-
.../apache/gravitino/policy/PolicyContents.java | 131 +++++++-
.../gravitino/policy/PolicyExecutionSpec.java | 283 ++++++++++++++++
.../gravitino/policy/TestPolicyContents.java | 65 ++++
.../gravitino/policy/TestPolicyExecutionSpec.java | 103 ++++++
.../json/ExecutablePolicyContentSerde.java | 370 +++++++++++++++++++++
.../java/org/apache/gravitino/json/JsonUtils.java | 49 ++-
.../storage/relational/utils/TestPOConverters.java | 139 ++++++++
design-docs/table-maintenance-service-design.md | 92 +++--
15 files changed, 1450 insertions(+), 87 deletions(-)
diff --git
a/api/src/main/java/org/apache/gravitino/policy/AbstractExecutablePolicyContent.java
b/api/src/main/java/org/apache/gravitino/policy/AbstractExecutablePolicyContent.java
new file mode 100644
index 0000000000..c24a827699
--- /dev/null
+++
b/api/src/main/java/org/apache/gravitino/policy/AbstractExecutablePolicyContent.java
@@ -0,0 +1,112 @@
+/*
+ * 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.policy;
+
+import java.util.Objects;
+import javax.annotation.Nullable;
+
+/**
+ * Base for built-in Iceberg maintenance policy contents that carry optional
execution peers.
+ *
+ * <p>Peers are supplied at construction as an immutable {@link
PolicyExecutionSpec}. Subclasses
+ * keep strategy fields separate and pass the spec through {@link
+ * #AbstractExecutablePolicyContent(PolicyExecutionSpec)}. Wire and
persistence JSON keep peer keys
+ * flat; mapping to and from those keys is handled in {@code common} (custom
Jackson serdes), not by
+ * flattening the five fields onto this type.
+ */
+public abstract class AbstractExecutablePolicyContent implements
ExecutablePolicyContent {
+
+ private final PolicyExecutionSpec executionSpec;
+
+ /**
+ * Creates content with the given execution peers.
+ *
+ * @param executionSpec immutable peers; {@code null} is treated as {@link
+ * PolicyExecutionSpec#empty()}
+ */
+ protected AbstractExecutablePolicyContent(@Nullable PolicyExecutionSpec
executionSpec) {
+ this.executionSpec = executionSpec == null ? PolicyExecutionSpec.empty() :
executionSpec;
+ }
+
+ /** Creates content with unset execution peers. */
+ protected AbstractExecutablePolicyContent() {
+ this(PolicyExecutionSpec.empty());
+ }
+
+ @Override
+ public PolicyExecutionSpec executionSpec() {
+ return executionSpec;
+ }
+
+ /**
+ * Validates universal {@link PolicyContent} rules plus shared
execution-peer checks.
+ *
+ * <p>Does not enforce type-specific maintenance contracts (for example
orphan cleanup forbidding
+ * {@code commitTrigger}, or cron parsing). Those belong on concrete
subclasses or the server
+ * create/alter path with the real {@link Policy.BuiltInType}.
+ */
+ @Override
+ public void validate() throws IllegalArgumentException {
+ ExecutablePolicyContent.super.validate();
+ executionSpec().validate();
+ }
+
+ /**
+ * Whether execution fields match.
+ *
+ * @param that other content
+ * @return true when execution fields are equal
+ */
+ protected boolean executionFieldsEqual(AbstractExecutablePolicyContent that)
{
+ return Objects.equals(executionSpec(), that.executionSpec());
+ }
+
+ /**
+ * @return hash of execution fields
+ */
+ protected int executionFieldsHashCode() {
+ return Objects.hashCode(executionSpec());
+ }
+
+ /**
+ * Appends execution fields to a subclass {@code toString} body.
+ *
+ * <p>Option map <em>values</em> are never rendered (keys only) so
accidental credentials are not
+ * exposed through logs or exception messages.
+ *
+ * <p>Callers that already wrote subclass fields should append {@code ", "}
before invoking this
+ * method.
+ *
+ * @param sb builder already containing an opening brace and any subclass
fields
+ */
+ protected void appendExecutionFields(StringBuilder sb) {
+ PolicyExecutionSpec spec = executionSpec();
+ sb.append("timeTrigger='")
+ .append(spec.timeTrigger())
+ .append('\'')
+ .append(", commitTrigger=")
+ .append(spec.commitTrigger())
+ .append(", minIntervalMs=")
+ .append(spec.minIntervalMs())
+ .append(", jobOptionsKeys=")
+ .append(spec.jobOptions().keySet())
+ .append(", updaterOptionsKeys=")
+ .append(spec.updaterOptions().keySet());
+ }
+}
diff --git
a/api/src/main/java/org/apache/gravitino/policy/ExecutablePolicyContent.java
b/api/src/main/java/org/apache/gravitino/policy/ExecutablePolicyContent.java
new file mode 100644
index 0000000000..7a6fa84296
--- /dev/null
+++ b/api/src/main/java/org/apache/gravitino/policy/ExecutablePolicyContent.java
@@ -0,0 +1,35 @@
+/*
+ * 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.policy;
+
+/**
+ * Policy content that carries immutable execution peers for automated
maintenance.
+ *
+ * <p>Row-filter, column-mask, and custom contents are not executable in this
sense. Callers that
+ * need peers should use {@code instanceof ExecutablePolicyContent} (or a
concrete maintenance
+ * content type) and read {@link #executionSpec()} rather than assuming every
{@link PolicyContent}
+ * exposes triggers or option bags.
+ */
+public interface ExecutablePolicyContent extends PolicyContent {
+
+ /**
+ * @return immutable execution peers; never {@code null}
+ */
+ PolicyExecutionSpec executionSpec();
+}
diff --git
a/api/src/main/java/org/apache/gravitino/policy/IcebergDataCompactionContent.java
b/api/src/main/java/org/apache/gravitino/policy/IcebergDataCompactionContent.java
index 08c8478d66..00aec6a0ec 100644
---
a/api/src/main/java/org/apache/gravitino/policy/IcebergDataCompactionContent.java
+++
b/api/src/main/java/org/apache/gravitino/policy/IcebergDataCompactionContent.java
@@ -32,7 +32,7 @@ import org.apache.commons.lang3.StringUtils;
import org.apache.gravitino.MetadataObject;
/** Built-in policy content for Iceberg compaction strategy. */
-public class IcebergDataCompactionContent implements PolicyContent {
+public class IcebergDataCompactionContent extends
AbstractExecutablePolicyContent {
/** Property key for strategy type. */
public static final String STRATEGY_TYPE_KEY = "strategy.type";
/** Strategy type value for iceberg data compaction. */
@@ -130,7 +130,7 @@ public class IcebergDataCompactionContent implements
PolicyContent {
/** Default constructor for Jackson deserialization only. */
private IcebergDataCompactionContent() {
- this(null, null, null, null, null, null, null, null);
+ this(null, null, null, null, null, null, null, null,
PolicyExecutionSpec.empty());
}
IcebergDataCompactionContent(
@@ -142,6 +142,29 @@ public class IcebergDataCompactionContent implements
PolicyContent {
String rewriteStrategy,
String sortOrder,
Map<String, String> rewriteOptions) {
+ this(
+ minDataFileMse,
+ minDeleteFileNumber,
+ dataFileMseWeight,
+ deleteFileNumberWeight,
+ maxPartitionNum,
+ rewriteStrategy,
+ sortOrder,
+ rewriteOptions,
+ PolicyExecutionSpec.empty());
+ }
+
+ IcebergDataCompactionContent(
+ Long minDataFileMse,
+ Long minDeleteFileNumber,
+ Long dataFileMseWeight,
+ Long deleteFileNumberWeight,
+ Long maxPartitionNum,
+ String rewriteStrategy,
+ String sortOrder,
+ Map<String, String> rewriteOptions,
+ PolicyExecutionSpec executionSpec) {
+ super(executionSpec);
// Nullable inputs are treated as "use default" to simplify policy
creation.
this.minDataFileMse = minDataFileMse == null ? DEFAULT_MIN_DATA_FILE_MSE :
minDataFileMse;
this.minDeleteFileNumber =
@@ -267,7 +290,7 @@ public class IcebergDataCompactionContent implements
PolicyContent {
@Override
public void validate() throws IllegalArgumentException {
- PolicyContent.super.validate();
+ super.validate();
// All fields are defaulted in the constructor, so only range checks are
needed here.
Preconditions.checkArgument(minDataFileMse >= 0, "minDataFileMse must be
>= 0");
Preconditions.checkArgument(minDeleteFileNumber >= 0, "minDeleteFileNumber
must be >= 0");
@@ -314,7 +337,8 @@ public class IcebergDataCompactionContent implements
PolicyContent {
return false;
}
IcebergDataCompactionContent that = (IcebergDataCompactionContent) o;
- return Objects.equals(minDataFileMse, that.minDataFileMse)
+ return executionFieldsEqual(that)
+ && Objects.equals(minDataFileMse, that.minDataFileMse)
&& Objects.equals(minDeleteFileNumber, that.minDeleteFileNumber)
&& Objects.equals(dataFileMseWeight, that.dataFileMseWeight)
&& Objects.equals(deleteFileNumberWeight, that.deleteFileNumberWeight)
@@ -327,6 +351,7 @@ public class IcebergDataCompactionContent implements
PolicyContent {
@Override
public int hashCode() {
return Objects.hash(
+ executionFieldsHashCode(),
minDataFileMse,
minDeleteFileNumber,
dataFileMseWeight,
@@ -339,25 +364,30 @@ public class IcebergDataCompactionContent implements
PolicyContent {
@Override
public String toString() {
- return "IcebergDataCompactionContent{"
- + "minDataFileMse="
- + minDataFileMse
- + ", minDeleteFileNumber="
- + minDeleteFileNumber
- + ", dataFileMseWeight="
- + dataFileMseWeight
- + ", deleteFileNumberWeight="
- + deleteFileNumberWeight
- + ", maxPartitionNum="
- + maxPartitionNum
- + ", rewriteStrategy='"
- + rewriteStrategy
- + '\''
- + ", sortOrder='"
- + sortOrder
- + '\''
- + ", rewriteOptions="
- + rewriteOptions
- + '}';
+ StringBuilder sb =
+ new StringBuilder("IcebergDataCompactionContent{")
+ .append("minDataFileMse=")
+ .append(minDataFileMse)
+ .append(", minDeleteFileNumber=")
+ .append(minDeleteFileNumber)
+ .append(", dataFileMseWeight=")
+ .append(dataFileMseWeight)
+ .append(", deleteFileNumberWeight=")
+ .append(deleteFileNumberWeight)
+ .append(", maxPartitionNum=")
+ .append(maxPartitionNum)
+ .append(", rewriteStrategy='")
+ .append(rewriteStrategy)
+ .append('\'')
+ .append(", sortOrder='")
+ .append(sortOrder)
+ .append('\'')
+ // Never render rewriteOptions values; same redaction as
jobOptions / updaterOptions.
+ .append(", rewriteOptionsKeys=")
+ .append(rewriteOptions.keySet())
+ .append(", ");
+ appendExecutionFields(sb);
+ sb.append('}');
+ return sb.toString();
}
}
diff --git
a/api/src/main/java/org/apache/gravitino/policy/IcebergOrphanFileRemovalContent.java
b/api/src/main/java/org/apache/gravitino/policy/IcebergOrphanFileRemovalContent.java
index 2823901b7b..441fe83516 100644
---
a/api/src/main/java/org/apache/gravitino/policy/IcebergOrphanFileRemovalContent.java
+++
b/api/src/main/java/org/apache/gravitino/policy/IcebergOrphanFileRemovalContent.java
@@ -31,7 +31,7 @@ import org.apache.commons.lang3.StringUtils;
import org.apache.gravitino.MetadataObject;
/** Configuration for table-level Iceberg orphan cleanup on an optimizer
invocation. */
-public class IcebergOrphanFileRemovalContent implements PolicyContent {
+public class IcebergOrphanFileRemovalContent extends
AbstractExecutablePolicyContent {
/** Property key for strategy type. */
public static final String STRATEGY_TYPE_KEY = "strategy.type";
/** Property key for the job template. */
@@ -64,10 +64,19 @@ public class IcebergOrphanFileRemovalContent implements
PolicyContent {
private final boolean dryRun;
private IcebergOrphanFileRemovalContent() {
- this(DEFAULT_OLDER_THAN_DAYS, null, DEFAULT_DRY_RUN);
+ this(DEFAULT_OLDER_THAN_DAYS, null, DEFAULT_DRY_RUN,
PolicyExecutionSpec.empty());
}
IcebergOrphanFileRemovalContent(long olderThanDays, @Nullable String
location, boolean dryRun) {
+ this(olderThanDays, location, dryRun, PolicyExecutionSpec.empty());
+ }
+
+ IcebergOrphanFileRemovalContent(
+ long olderThanDays,
+ @Nullable String location,
+ boolean dryRun,
+ @Nullable PolicyExecutionSpec executionSpec) {
+ super(executionSpec);
this.olderThanDays = olderThanDays;
this.location = location;
this.dryRun = dryRun;
@@ -136,7 +145,10 @@ public class IcebergOrphanFileRemovalContent implements
PolicyContent {
@Override
public void validate() {
- PolicyContent.super.validate();
+ super.validate();
+ Preconditions.checkArgument(
+ !Boolean.TRUE.equals(executionSpec().commitTrigger()),
+ "commitTrigger is not allowed for orphan file removal");
validateOlderThanDays(olderThanDays);
Preconditions.checkArgument(
location == null || StringUtils.isNotBlank(location), "location must
not be blank");
@@ -151,13 +163,14 @@ public class IcebergOrphanFileRemovalContent implements
PolicyContent {
return false;
}
IcebergOrphanFileRemovalContent that = (IcebergOrphanFileRemovalContent)
other;
- return olderThanDays == that.olderThanDays
+ return executionFieldsEqual(that)
+ && olderThanDays == that.olderThanDays
&& dryRun == that.dryRun
&& Objects.equals(location, that.location);
}
@Override
public int hashCode() {
- return Objects.hash(olderThanDays, location, dryRun);
+ return Objects.hash(executionFieldsHashCode(), olderThanDays, location,
dryRun);
}
}
diff --git
a/api/src/main/java/org/apache/gravitino/policy/IcebergRewriteManifestsContent.java
b/api/src/main/java/org/apache/gravitino/policy/IcebergRewriteManifestsContent.java
index 53d59202cb..0c1fa54d41 100644
---
a/api/src/main/java/org/apache/gravitino/policy/IcebergRewriteManifestsContent.java
+++
b/api/src/main/java/org/apache/gravitino/policy/IcebergRewriteManifestsContent.java
@@ -24,15 +24,16 @@ import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Set;
+import javax.annotation.Nullable;
import org.apache.gravitino.MetadataObject;
/**
* Placeholder built-in policy content for Iceberg manifest rewrite.
*
* <p>Configurable thresholds will be added when the rewrite-manifests policy
lands; this class
- * registers the built-in type for TMS task mapping.
+ * registers {@link Policy.BuiltInType#ICEBERG_REWRITE_MANIFESTS}.
*/
-public class IcebergRewriteManifestsContent implements PolicyContent {
+public class IcebergRewriteManifestsContent extends
AbstractExecutablePolicyContent {
/** Property key for strategy type. */
public static final String STRATEGY_TYPE_KEY = "strategy.type";
/** Property key for the job template. */
@@ -51,7 +52,13 @@ public class IcebergRewriteManifestsContent implements
PolicyContent {
MetadataObject.Type.CATALOG, MetadataObject.Type.SCHEMA,
MetadataObject.Type.TABLE);
/** Default constructor for Jackson deserialization only. */
- public IcebergRewriteManifestsContent() {}
+ public IcebergRewriteManifestsContent() {
+ this(PolicyExecutionSpec.empty());
+ }
+
+ IcebergRewriteManifestsContent(@Nullable PolicyExecutionSpec executionSpec) {
+ super(executionSpec);
+ }
@Override
public Set<MetadataObject.Type> supportedObjectTypes() {
@@ -75,16 +82,20 @@ public class IcebergRewriteManifestsContent implements
PolicyContent {
@Override
public boolean equals(Object other) {
- return other instanceof IcebergRewriteManifestsContent;
+ return other instanceof IcebergRewriteManifestsContent
+ && executionFieldsEqual((IcebergRewriteManifestsContent) other);
}
@Override
public int hashCode() {
- return IcebergRewriteManifestsContent.class.hashCode();
+ return executionFieldsHashCode();
}
@Override
public String toString() {
- return "IcebergRewriteManifestsContent{}";
+ StringBuilder sb = new StringBuilder("IcebergRewriteManifestsContent{");
+ appendExecutionFields(sb);
+ sb.append('}');
+ return sb.toString();
}
}
diff --git
a/api/src/main/java/org/apache/gravitino/policy/IcebergSnapshotExpirationContent.java
b/api/src/main/java/org/apache/gravitino/policy/IcebergSnapshotExpirationContent.java
index d752d8f905..0d4071f25e 100644
---
a/api/src/main/java/org/apache/gravitino/policy/IcebergSnapshotExpirationContent.java
+++
b/api/src/main/java/org/apache/gravitino/policy/IcebergSnapshotExpirationContent.java
@@ -24,15 +24,16 @@ import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Set;
+import javax.annotation.Nullable;
import org.apache.gravitino.MetadataObject;
/**
* Placeholder built-in policy content for Iceberg snapshot expiration.
*
* <p>Configurable thresholds will be added when the expire-snapshots policy
lands; this class
- * registers the built-in type for TMS task mapping.
+ * registers {@link Policy.BuiltInType#ICEBERG_SNAPSHOT_EXPIRATION}.
*/
-public class IcebergSnapshotExpirationContent implements PolicyContent {
+public class IcebergSnapshotExpirationContent extends
AbstractExecutablePolicyContent {
/** Property key for strategy type. */
public static final String STRATEGY_TYPE_KEY = "strategy.type";
/** Property key for the job template. */
@@ -51,7 +52,13 @@ public class IcebergSnapshotExpirationContent implements
PolicyContent {
MetadataObject.Type.CATALOG, MetadataObject.Type.SCHEMA,
MetadataObject.Type.TABLE);
/** Default constructor for Jackson deserialization only. */
- public IcebergSnapshotExpirationContent() {}
+ public IcebergSnapshotExpirationContent() {
+ this(PolicyExecutionSpec.empty());
+ }
+
+ IcebergSnapshotExpirationContent(@Nullable PolicyExecutionSpec
executionSpec) {
+ super(executionSpec);
+ }
@Override
public Set<MetadataObject.Type> supportedObjectTypes() {
@@ -75,16 +82,20 @@ public class IcebergSnapshotExpirationContent implements
PolicyContent {
@Override
public boolean equals(Object other) {
- return other instanceof IcebergSnapshotExpirationContent;
+ return other instanceof IcebergSnapshotExpirationContent
+ && executionFieldsEqual((IcebergSnapshotExpirationContent) other);
}
@Override
public int hashCode() {
- return IcebergSnapshotExpirationContent.class.hashCode();
+ return executionFieldsHashCode();
}
@Override
public String toString() {
- return "IcebergSnapshotExpirationContent{}";
+ StringBuilder sb = new StringBuilder("IcebergSnapshotExpirationContent{");
+ appendExecutionFields(sb);
+ sb.append('}');
+ return sb.toString();
}
}
diff --git a/api/src/main/java/org/apache/gravitino/policy/PolicyContent.java
b/api/src/main/java/org/apache/gravitino/policy/PolicyContent.java
index 03f6168679..dfe48d9e50 100644
--- a/api/src/main/java/org/apache/gravitino/policy/PolicyContent.java
+++ b/api/src/main/java/org/apache/gravitino/policy/PolicyContent.java
@@ -24,7 +24,13 @@ import java.util.Set;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.gravitino.MetadataObject;
-/** The interface of the content of the policy. */
+/**
+ * The interface of the content of the policy.
+ *
+ * <p>Execution peers ({@code timeTrigger}, {@code commitTrigger}, {@code
minIntervalMs}, {@code
+ * jobOptions}, {@code updaterOptions}) are not part of this universal
interface. Built-in Iceberg
+ * maintenance contents carry them via {@link ExecutablePolicyContent}/{@link
PolicyExecutionSpec}.
+ */
public interface PolicyContent {
/**
@@ -50,6 +56,9 @@ public interface PolicyContent {
/**
* Validates the policy content.
*
+ * <p>Checks universal fields such as {@link #supportedObjectTypes()}.
Maintenance execution-field
+ * validation lives on {@link ExecutablePolicyContent} implementations.
+ *
* @throws IllegalArgumentException if the content is invalid.
*/
default void validate() throws IllegalArgumentException {
diff --git a/api/src/main/java/org/apache/gravitino/policy/PolicyContents.java
b/api/src/main/java/org/apache/gravitino/policy/PolicyContents.java
index b7e92193f6..4eba5b102a 100644
--- a/api/src/main/java/org/apache/gravitino/policy/PolicyContents.java
+++ b/api/src/main/java/org/apache/gravitino/policy/PolicyContents.java
@@ -134,6 +134,55 @@ public class PolicyContents {
String rewriteStrategy,
String sortOrder,
Map<String, String> rewriteOptions) {
+ return icebergDataCompaction(
+ minDataFileMse,
+ minDeleteFileNumber,
+ dataFileMseWeight,
+ deleteFileNumberWeight,
+ maxPartitionNum,
+ rewriteStrategy,
+ sortOrder,
+ rewriteOptions,
+ null,
+ null,
+ null,
+ null,
+ null);
+ }
+
+ /**
+ * Creates an iceberg compaction policy content with rewrite strategy, sort
order, and execution
+ * peers.
+ *
+ * @param minDataFileMse minimum threshold for custom-data-file-mse
+ * @param minDeleteFileNumber minimum threshold for custom-delete-file-number
+ * @param dataFileMseWeight weight used for custom-data-file-mse score
contribution
+ * @param deleteFileNumberWeight weight used for custom-delete-file-number
score contribution
+ * @param maxPartitionNum maximum partition number selected for compaction
+ * @param rewriteStrategy Iceberg rewrite_data_files top-level strategy
+ * @param sortOrder Iceberg rewrite_data_files top-level sort_order
+ * @param rewriteOptions rewrite options forwarded as job.options.*
+ * @param timeTrigger optional cron expression
+ * @param commitTrigger optional commit-trigger flag
+ * @param minIntervalMs optional cooldown ms
+ * @param jobOptions optional job options
+ * @param updaterOptions optional updater options
+ * @return iceberg compaction policy content
+ */
+ public static PolicyContent icebergDataCompaction(
+ long minDataFileMse,
+ long minDeleteFileNumber,
+ long dataFileMseWeight,
+ long deleteFileNumberWeight,
+ long maxPartitionNum,
+ String rewriteStrategy,
+ String sortOrder,
+ Map<String, String> rewriteOptions,
+ @Nullable String timeTrigger,
+ @Nullable Boolean commitTrigger,
+ @Nullable Long minIntervalMs,
+ @Nullable Map<String, String> jobOptions,
+ @Nullable Map<String, String> updaterOptions) {
return new IcebergDataCompactionContent(
minDataFileMse,
minDeleteFileNumber,
@@ -142,7 +191,9 @@ public class PolicyContents {
maxPartitionNum,
rewriteStrategy,
sortOrder,
- rewriteOptions);
+ rewriteOptions,
+ PolicyExecutionSpec.of(
+ timeTrigger, commitTrigger, minIntervalMs, jobOptions,
updaterOptions));
}
/**
@@ -165,7 +216,37 @@ public class PolicyContents {
*/
public static PolicyContent icebergOrphanFileRemoval(
long olderThanDays, @Nullable String location, boolean dryRun) {
- return new IcebergOrphanFileRemovalContent(olderThanDays, location,
dryRun);
+ return icebergOrphanFileRemoval(olderThanDays, location, dryRun, null,
null, null, null, null);
+ }
+
+ /**
+ * Creates orphan cleanup policy content with execution peers.
+ *
+ * @param olderThanDays minimum file age in days, from 1 to 36500 inclusive
+ * @param location optional scan location within the table
+ * @param dryRun whether to list candidates without deleting them
+ * @param timeTrigger optional cron expression
+ * @param commitTrigger optional commit-trigger flag (must not be {@code
true})
+ * @param minIntervalMs optional cooldown ms
+ * @param jobOptions optional job options
+ * @param updaterOptions optional updater options
+ * @return orphan cleanup policy content
+ */
+ public static PolicyContent icebergOrphanFileRemoval(
+ long olderThanDays,
+ @Nullable String location,
+ boolean dryRun,
+ @Nullable String timeTrigger,
+ @Nullable Boolean commitTrigger,
+ @Nullable Long minIntervalMs,
+ @Nullable Map<String, String> jobOptions,
+ @Nullable Map<String, String> updaterOptions) {
+ return new IcebergOrphanFileRemovalContent(
+ olderThanDays,
+ location,
+ dryRun,
+ PolicyExecutionSpec.of(
+ timeTrigger, commitTrigger, minIntervalMs, jobOptions,
updaterOptions));
}
/**
@@ -174,7 +255,28 @@ public class PolicyContents {
* @return rewrite-manifests policy content
*/
public static PolicyContent icebergRewriteManifests() {
- return new IcebergRewriteManifestsContent();
+ return icebergRewriteManifests(null, null, null, null, null);
+ }
+
+ /**
+ * Creates placeholder Iceberg rewrite-manifests policy content with
execution peers.
+ *
+ * @param timeTrigger optional cron expression
+ * @param commitTrigger optional commit-trigger flag
+ * @param minIntervalMs optional cooldown ms
+ * @param jobOptions optional job options
+ * @param updaterOptions optional updater options
+ * @return rewrite-manifests policy content
+ */
+ public static PolicyContent icebergRewriteManifests(
+ @Nullable String timeTrigger,
+ @Nullable Boolean commitTrigger,
+ @Nullable Long minIntervalMs,
+ @Nullable Map<String, String> jobOptions,
+ @Nullable Map<String, String> updaterOptions) {
+ return new IcebergRewriteManifestsContent(
+ PolicyExecutionSpec.of(
+ timeTrigger, commitTrigger, minIntervalMs, jobOptions,
updaterOptions));
}
/**
@@ -183,7 +285,28 @@ public class PolicyContents {
* @return snapshot-expiration policy content
*/
public static PolicyContent icebergSnapshotExpiration() {
- return new IcebergSnapshotExpirationContent();
+ return icebergSnapshotExpiration(null, null, null, null, null);
+ }
+
+ /**
+ * Creates placeholder Iceberg snapshot-expiration policy content with
execution peers.
+ *
+ * @param timeTrigger optional cron expression
+ * @param commitTrigger optional commit-trigger flag
+ * @param minIntervalMs optional cooldown ms
+ * @param jobOptions optional job options
+ * @param updaterOptions optional updater options
+ * @return snapshot-expiration policy content
+ */
+ public static PolicyContent icebergSnapshotExpiration(
+ @Nullable String timeTrigger,
+ @Nullable Boolean commitTrigger,
+ @Nullable Long minIntervalMs,
+ @Nullable Map<String, String> jobOptions,
+ @Nullable Map<String, String> updaterOptions) {
+ return new IcebergSnapshotExpirationContent(
+ PolicyExecutionSpec.of(
+ timeTrigger, commitTrigger, minIntervalMs, jobOptions,
updaterOptions));
}
/**
diff --git
a/api/src/main/java/org/apache/gravitino/policy/PolicyExecutionSpec.java
b/api/src/main/java/org/apache/gravitino/policy/PolicyExecutionSpec.java
new file mode 100644
index 0000000000..24ab7ba572
--- /dev/null
+++ b/api/src/main/java/org/apache/gravitino/policy/PolicyExecutionSpec.java
@@ -0,0 +1,283 @@
+/*
+ * 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.policy;
+
+import com.google.common.base.Preconditions;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Objects;
+import javax.annotation.Nullable;
+import org.apache.commons.lang3.StringUtils;
+
+/**
+ * Immutable execution peer fields for executable policy contents: triggers,
cooldown, and option
+ * bags.
+ *
+ * <p>This type is Jackson-independent in-memory composition for {@link
ExecutablePolicyContent}.
+ * Wire and persistence JSON keep the flat peer keys ({@code timeTrigger},
{@code commitTrigger},
+ * {@code minIntervalMs}, {@code jobOptions}, {@code updaterOptions}); {@code
common} maps those
+ * keys to and from this value object via custom Jackson serdes (same idea as
{@code
+ * DistributionSerializer}).
+ *
+ * <p>Invoked from {@link AbstractExecutablePolicyContent}, not from universal
{@link
+ * PolicyContent#validate()}. Type-specific rules (for example rejecting
{@code commitTrigger} for
+ * orphan cleanup) belong on the concrete content or the server create/alter
path.
+ *
+ * <p>{@code timeTrigger} cron syntax is <strong>not</strong> validated here.
The scheduler
+ * registration path (db-scheduler / runtime cron parser) is the source of
truth; validating with a
+ * regex or a second library in {@code api} would drift from that runtime and
pull scheduler
+ * dependencies into the public API module.
+ */
+public final class PolicyExecutionSpec {
+
+ /** Spec with every peer unset. */
+ public static final PolicyExecutionSpec EMPTY =
+ new PolicyExecutionSpec(null, null, null, null, null);
+
+ @Nullable private final String timeTrigger;
+ @Nullable private final Boolean commitTrigger;
+ @Nullable private final Long minIntervalMs;
+ private final Map<String, String> jobOptions;
+ private final Map<String, String> updaterOptions;
+
+ private PolicyExecutionSpec(
+ @Nullable String timeTrigger,
+ @Nullable Boolean commitTrigger,
+ @Nullable Long minIntervalMs,
+ @Nullable Map<String, String> jobOptions,
+ @Nullable Map<String, String> updaterOptions) {
+ this.timeTrigger = StringUtils.isBlank(timeTrigger) ? null : timeTrigger;
+ this.commitTrigger = commitTrigger;
+ this.minIntervalMs = minIntervalMs;
+ this.jobOptions = copyOptions(jobOptions);
+ this.updaterOptions = copyOptions(updaterOptions);
+ }
+
+ /**
+ * Creates an immutable execution spec.
+ *
+ * @param timeTrigger optional cron expression
+ * @param commitTrigger optional commit-trigger flag; any non-null value
(including {@code false})
+ * is retained
+ * @param minIntervalMs optional cooldown ms
+ * @param jobOptions optional job options
+ * @param updaterOptions optional updater options
+ * @return immutable spec, or {@link #EMPTY} when every peer is unset
+ */
+ public static PolicyExecutionSpec of(
+ @Nullable String timeTrigger,
+ @Nullable Boolean commitTrigger,
+ @Nullable Long minIntervalMs,
+ @Nullable Map<String, String> jobOptions,
+ @Nullable Map<String, String> updaterOptions) {
+ if (!hasAny(timeTrigger, commitTrigger, minIntervalMs, jobOptions,
updaterOptions)) {
+ return EMPTY;
+ }
+ return new PolicyExecutionSpec(
+ timeTrigger, commitTrigger, minIntervalMs, jobOptions, updaterOptions);
+ }
+
+ /**
+ * @return empty spec
+ */
+ public static PolicyExecutionSpec empty() {
+ return EMPTY;
+ }
+
+ /**
+ * @return cron expression, or {@code null} when unset
+ */
+ @Nullable
+ public String timeTrigger() {
+ return timeTrigger;
+ }
+
+ /**
+ * @return commit-trigger flag when explicitly supplied (including {@code
false}), or {@code null}
+ * when unset
+ */
+ @Nullable
+ public Boolean commitTrigger() {
+ return commitTrigger;
+ }
+
+ /**
+ * @return minimum interval in milliseconds, or {@code null} when unset
+ */
+ @Nullable
+ public Long minIntervalMs() {
+ return minIntervalMs;
+ }
+
+ /**
+ * @return job options; never {@code null}
+ */
+ public Map<String, String> jobOptions() {
+ return jobOptions;
+ }
+
+ /**
+ * @return updater options; never {@code null}
+ */
+ public Map<String, String> updaterOptions() {
+ return updaterOptions;
+ }
+
+ /**
+ * @return {@code true} when any peer is set
+ */
+ public boolean hasAny() {
+ return hasAny(timeTrigger, commitTrigger, minIntervalMs, jobOptions,
updaterOptions);
+ }
+
+ /**
+ * @param timeTrigger replacement cron expression
+ * @return new spec with updated {@code timeTrigger}
+ */
+ public PolicyExecutionSpec withTimeTrigger(@Nullable String timeTrigger) {
+ return of(timeTrigger, commitTrigger, minIntervalMs, jobOptions,
updaterOptions);
+ }
+
+ /**
+ * @param commitTrigger replacement commit-trigger flag
+ * @return new spec with updated {@code commitTrigger}
+ */
+ public PolicyExecutionSpec withCommitTrigger(@Nullable Boolean
commitTrigger) {
+ return of(timeTrigger, commitTrigger, minIntervalMs, jobOptions,
updaterOptions);
+ }
+
+ /**
+ * @param minIntervalMs replacement cooldown
+ * @return new spec with updated {@code minIntervalMs}
+ */
+ public PolicyExecutionSpec withMinIntervalMs(@Nullable Long minIntervalMs) {
+ return of(timeTrigger, commitTrigger, minIntervalMs, jobOptions,
updaterOptions);
+ }
+
+ /**
+ * @param jobOptions replacement job options
+ * @return new spec with updated {@code jobOptions}
+ */
+ public PolicyExecutionSpec withJobOptions(@Nullable Map<String, String>
jobOptions) {
+ return of(timeTrigger, commitTrigger, minIntervalMs, jobOptions,
updaterOptions);
+ }
+
+ /**
+ * @param updaterOptions replacement updater options
+ * @return new spec with updated {@code updaterOptions}
+ */
+ public PolicyExecutionSpec withUpdaterOptions(@Nullable Map<String, String>
updaterOptions) {
+ return of(timeTrigger, commitTrigger, minIntervalMs, jobOptions,
updaterOptions);
+ }
+
+ /**
+ * Validates shared peer constraints (cooldown and option maps).
+ *
+ * <p>Does not parse {@code timeTrigger}; cron validity is checked when the
expression is
+ * registered with the scheduler.
+ */
+ public void validate() {
+ if (!hasAny()) {
+ return;
+ }
+ if (minIntervalMs != null) {
+ Preconditions.checkArgument(minIntervalMs >= 0, "minIntervalMs must be
>= 0");
+ }
+ validateOptionMap(jobOptions, "jobOptions");
+ validateOptionMap(updaterOptions, "updaterOptions");
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ if (this == other) {
+ return true;
+ }
+ if (!(other instanceof PolicyExecutionSpec)) {
+ return false;
+ }
+ PolicyExecutionSpec that = (PolicyExecutionSpec) other;
+ return Objects.equals(timeTrigger, that.timeTrigger)
+ && Objects.equals(commitTrigger, that.commitTrigger)
+ && Objects.equals(minIntervalMs, that.minIntervalMs)
+ && Objects.equals(jobOptions, that.jobOptions)
+ && Objects.equals(updaterOptions, that.updaterOptions);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(timeTrigger, commitTrigger, minIntervalMs, jobOptions,
updaterOptions);
+ }
+
+ @Override
+ public String toString() {
+ // Never render option map values; credentials may appear before
sensitive-key rejection.
+ return "PolicyExecutionSpec{"
+ + "timeTrigger='"
+ + timeTrigger
+ + '\''
+ + ", commitTrigger="
+ + commitTrigger
+ + ", minIntervalMs="
+ + minIntervalMs
+ + ", jobOptionsKeys="
+ + jobOptions.keySet()
+ + ", updaterOptionsKeys="
+ + updaterOptions.keySet()
+ + '}';
+ }
+
+ /**
+ * Whether any execution peer field is set.
+ *
+ * <p>Every non-null {@code commitTrigger} (including {@code false}) counts
as explicitly
+ * supplied, so JSON round-trips and {@code equals}/{@code hashCode} do not
depend on other peers.
+ */
+ private static boolean hasAny(
+ @Nullable String timeTrigger,
+ @Nullable Boolean commitTrigger,
+ @Nullable Long minIntervalMs,
+ @Nullable Map<String, String> jobOptions,
+ @Nullable Map<String, String> updaterOptions) {
+ return StringUtils.isNotBlank(timeTrigger)
+ || commitTrigger != null
+ || minIntervalMs != null
+ || (jobOptions != null && !jobOptions.isEmpty())
+ || (updaterOptions != null && !updaterOptions.isEmpty());
+ }
+
+ private static Map<String, String> copyOptions(@Nullable Map<String, String>
options) {
+ if (options == null || options.isEmpty()) {
+ return Collections.emptyMap();
+ }
+ return Collections.unmodifiableMap(new LinkedHashMap<>(options));
+ }
+
+ private static void validateOptionMap(Map<String, String> options, String
mapName) {
+ if (options == null || options.isEmpty()) {
+ return;
+ }
+ options.forEach(
+ (key, value) -> {
+ Preconditions.checkArgument(StringUtils.isNotBlank(key), "%s key is
blank", mapName);
+ Preconditions.checkArgument(
+ StringUtils.isNotBlank(value), "%s '%s' must have non-empty
value", mapName, key);
+ });
+ }
+}
diff --git
a/api/src/test/java/org/apache/gravitino/policy/TestPolicyContents.java
b/api/src/test/java/org/apache/gravitino/policy/TestPolicyContents.java
index 4bac0a9d0f..5d12087b5d 100644
--- a/api/src/test/java/org/apache/gravitino/policy/TestPolicyContents.java
+++ b/api/src/test/java/org/apache/gravitino/policy/TestPolicyContents.java
@@ -19,6 +19,7 @@
package org.apache.gravitino.policy;
+import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import java.util.HashMap;
import java.util.Map;
@@ -172,6 +173,70 @@ public class TestPolicyContents {
Assertions.assertTrue(exception.getMessage().contains("maxPartitionNum"));
}
+ @Test
+ void testValidateRejectsNegativeMinIntervalMsOnExecutableContent() {
+ PolicyContent content =
+ PolicyContents.icebergDataCompaction(
+ IcebergDataCompactionContent.DEFAULT_MIN_DATA_FILE_MSE,
+ IcebergDataCompactionContent.DEFAULT_MIN_DELETE_FILE_NUMBER,
+ IcebergDataCompactionContent.DEFAULT_DATA_FILE_MSE_WEIGHT,
+ IcebergDataCompactionContent.DEFAULT_DELETE_FILE_NUMBER_WEIGHT,
+ IcebergDataCompactionContent.DEFAULT_MAX_PARTITION_NUM,
+ IcebergDataCompactionContent.DEFAULT_REWRITE_STRATEGY,
+ IcebergDataCompactionContent.DEFAULT_SORT_ORDER,
+ null,
+ null,
+ null,
+ -1L,
+ null,
+ null);
+
+ IllegalArgumentException exception =
+ Assertions.assertThrows(IllegalArgumentException.class,
content::validate);
+ Assertions.assertTrue(exception.getMessage().contains("minIntervalMs"));
+ }
+
+ @Test
+ void testExecutionPeersOnlyOnExecutablePolicyContent() {
+ Assertions.assertInstanceOf(
+ ExecutablePolicyContent.class, PolicyContents.icebergDataCompaction());
+ Assertions.assertInstanceOf(
+ ExecutablePolicyContent.class,
PolicyContents.icebergOrphanFileRemoval());
+ Assertions.assertFalse(
+ PolicyContents.custom(
+ ImmutableMap.of("rule", "v"),
ImmutableSet.of(MetadataObject.Type.TABLE), null)
+ instanceof ExecutablePolicyContent);
+ }
+
+ @Test
+ void testValidateAcceptsBuiltInContentsWithoutExecutionPeers() {
+ Assertions.assertDoesNotThrow(() ->
PolicyContents.icebergDataCompaction().validate());
+ Assertions.assertDoesNotThrow(() ->
PolicyContents.icebergOrphanFileRemoval().validate());
+ Assertions.assertDoesNotThrow(
+ () ->
+ PolicyContents.custom(
+ ImmutableMap.of("rule", "v"),
ImmutableSet.of(MetadataObject.Type.TABLE), null)
+ .validate());
+ }
+
+ @Test
+ void testOrphanFileRemovalRejectsCommitTrigger() {
+ PolicyContent content =
+ PolicyContents.icebergOrphanFileRemoval(
+ IcebergOrphanFileRemovalContent.DEFAULT_OLDER_THAN_DAYS,
+ null,
+ IcebergOrphanFileRemovalContent.DEFAULT_DRY_RUN,
+ null,
+ true,
+ null,
+ null,
+ null);
+
+ IllegalArgumentException exception =
+ Assertions.assertThrows(IllegalArgumentException.class,
content::validate);
+ Assertions.assertTrue(exception.getMessage().contains("commitTrigger"));
+ }
+
private static Map<String, String> mapOf(String key, String value) {
Map<String, String> map = new HashMap<>();
map.put(key, value);
diff --git
a/api/src/test/java/org/apache/gravitino/policy/TestPolicyExecutionSpec.java
b/api/src/test/java/org/apache/gravitino/policy/TestPolicyExecutionSpec.java
new file mode 100644
index 0000000000..cf36a06d66
--- /dev/null
+++ b/api/src/test/java/org/apache/gravitino/policy/TestPolicyExecutionSpec.java
@@ -0,0 +1,103 @@
+/*
+ * 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.policy;
+
+import com.google.common.collect.ImmutableMap;
+import java.util.Collections;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TestPolicyExecutionSpec {
+
+ @Test
+ void emptyPeersReuseSingleton() {
+ Assertions.assertSame(
+ PolicyExecutionSpec.EMPTY, PolicyExecutionSpec.of(null, null, null,
null, null));
+ Assertions.assertSame(PolicyExecutionSpec.EMPTY,
PolicyExecutionSpec.empty());
+ }
+
+ @Test
+ void commitTriggerFalseIsRetainedWhenAlone() {
+ PolicyExecutionSpec spec = PolicyExecutionSpec.of(null, false, null, null,
null);
+ Assertions.assertNotSame(PolicyExecutionSpec.EMPTY, spec);
+ Assertions.assertEquals(Boolean.FALSE, spec.commitTrigger());
+ Assertions.assertTrue(spec.hasAny());
+ }
+
+ @Test
+ void toStringOmitsOptionValues() {
+ PolicyExecutionSpec spec =
+ PolicyExecutionSpec.of(
+ "0 0 2 * * ?",
+ true,
+ 1000L,
+ ImmutableMap.of("password", "secret"),
+ ImmutableMap.of("oauth_credential", "id:secret"));
+
+ String text = spec.toString();
+ Assertions.assertTrue(text.contains("jobOptionsKeys=[password]"));
+
Assertions.assertTrue(text.contains("updaterOptionsKeys=[oauth_credential]"));
+ Assertions.assertFalse(text.contains("secret"));
+ Assertions.assertFalse(text.contains("id:secret"));
+ }
+
+ @Test
+ void validateAcceptsScalarTriggers() {
+ PolicyExecutionSpec spec =
+ PolicyExecutionSpec.of(
+ "0 0 2 * * ?",
+ true,
+ 100L,
+ ImmutableMap.of("uri", "http://irc";),
+ ImmutableMap.of("metalake", "ml"));
+ Assertions.assertDoesNotThrow(spec::validate);
+ }
+
+ @Test
+ void validateAcceptsCommitTriggerTrueAlone() {
+ PolicyExecutionSpec spec =
+ PolicyExecutionSpec.of(null, true, null, Collections.emptyMap(),
Collections.emptyMap());
+ Assertions.assertDoesNotThrow(spec::validate);
+ }
+
+ @Test
+ void doesNotParseTimeTriggerCron() {
+ // Cron syntax is owned by the scheduler registration path, not this API
type.
+ PolicyExecutionSpec spec =
+ PolicyExecutionSpec.of(
+ "0 2 * *", null, null, Collections.emptyMap(),
Collections.emptyMap());
+ Assertions.assertDoesNotThrow(spec::validate);
+ }
+
+ @Test
+ void blankTimeTriggerTreatedAsUnset() {
+ Assertions.assertSame(
+ PolicyExecutionSpec.EMPTY,
+ PolicyExecutionSpec.of(" ", null, null, Collections.emptyMap(),
Collections.emptyMap()));
+ Assertions.assertDoesNotThrow(PolicyExecutionSpec.EMPTY::validate);
+ }
+
+ @Test
+ void validateRejectsNegativeMinInterval() {
+ PolicyExecutionSpec spec = PolicyExecutionSpec.of(null, null, -1L, null,
null);
+ IllegalArgumentException ex =
+ Assertions.assertThrows(IllegalArgumentException.class,
spec::validate);
+ Assertions.assertTrue(ex.getMessage().contains("minIntervalMs"));
+ }
+}
diff --git
a/common/src/main/java/org/apache/gravitino/json/ExecutablePolicyContentSerde.java
b/common/src/main/java/org/apache/gravitino/json/ExecutablePolicyContentSerde.java
new file mode 100644
index 0000000000..a033614e57
--- /dev/null
+++
b/common/src/main/java/org/apache/gravitino/json/ExecutablePolicyContentSerde.java
@@ -0,0 +1,370 @@
+/*
+ * 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.json;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.JsonDeserializer;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+import com.google.common.base.Preconditions;
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import javax.annotation.Nullable;
+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.PolicyContents;
+import org.apache.gravitino.policy.PolicyExecutionSpec;
+
+/**
+ * Jackson serdes for executable policy contents and {@link
PolicyExecutionSpec}.
+ *
+ * <p>Follows the same idea as {@link JsonUtils.DistributionSerializer}: keep
an immutable composed
+ * value in memory, and map it to or from the flat wire/persistence keys
({@code timeTrigger},
+ * {@code commitTrigger}, {@code minIntervalMs}, {@code jobOptions}, {@code
updaterOptions}) without
+ * nesting an {@code executionSpec} object.
+ */
+public final class ExecutablePolicyContentSerde {
+
+ private ExecutablePolicyContentSerde() {}
+
+ /**
+ * Registers serdes used when persisting domain {@link
org.apache.gravitino.policy.PolicyContent}
+ * via {@link JsonUtils#anyFieldMapper()}.
+ *
+ * @param module module to register on
+ */
+ public static void registerDomainSerdes(SimpleModule module) {
+ module.addSerializer(
+ IcebergDataCompactionContent.class, new
IcebergDataCompactionContentSerializer());
+ module.addDeserializer(
+ IcebergDataCompactionContent.class, new
IcebergDataCompactionContentDeserializer());
+ module.addSerializer(
+ IcebergOrphanFileRemovalContent.class, new
IcebergOrphanFileRemovalContentSerializer());
+ module.addDeserializer(
+ IcebergOrphanFileRemovalContent.class, new
IcebergOrphanFileRemovalContentDeserializer());
+ module.addSerializer(
+ IcebergRewriteManifestsContent.class, new
IcebergRewriteManifestsContentSerializer());
+ module.addDeserializer(
+ IcebergRewriteManifestsContent.class, new
IcebergRewriteManifestsContentDeserializer());
+ module.addSerializer(
+ IcebergSnapshotExpirationContent.class, new
IcebergSnapshotExpirationContentSerializer());
+ module.addDeserializer(
+ IcebergSnapshotExpirationContent.class, new
IcebergSnapshotExpirationContentDeserializer());
+ }
+
+ /**
+ * Writes execution peer fields as siblings of other content properties (no
object wrapper).
+ *
+ * @param spec execution peers; {@code null} or empty writes nothing
+ * @param gen JSON generator already inside a content object
+ * @throws IOException if writing JSON fails
+ */
+ public static void writeExecutionPeerFields(@Nullable PolicyExecutionSpec
spec, JsonGenerator gen)
+ throws IOException {
+ if (spec == null || !spec.hasAny()) {
+ return;
+ }
+ if (spec.timeTrigger() != null) {
+ gen.writeStringField("timeTrigger", spec.timeTrigger());
+ }
+ if (spec.commitTrigger() != null) {
+ gen.writeBooleanField("commitTrigger", spec.commitTrigger());
+ }
+ if (spec.minIntervalMs() != null) {
+ gen.writeNumberField("minIntervalMs", spec.minIntervalMs());
+ }
+ if (!spec.jobOptions().isEmpty()) {
+ gen.writeObjectField("jobOptions", spec.jobOptions());
+ }
+ if (!spec.updaterOptions().isEmpty()) {
+ gen.writeObjectField("updaterOptions", spec.updaterOptions());
+ }
+ }
+
+ /**
+ * Reads flat execution peer keys from a content JSON object.
+ *
+ * @param node content object node
+ * @return immutable spec (possibly {@link PolicyExecutionSpec#empty()})
+ */
+ public static PolicyExecutionSpec readExecutionSpec(JsonNode node) {
+ Preconditions.checkArgument(
+ node != null && node.isObject(),
+ "Cannot parse execution peers from invalid JSON: %s",
+ node);
+ String timeTrigger = textOrNull(node, "timeTrigger");
+ Boolean commitTrigger = booleanOrNull(node, "commitTrigger");
+ Long minIntervalMs = longOrNull(node, "minIntervalMs");
+ Map<String, String> jobOptions = stringMapOrNull(node, "jobOptions");
+ Map<String, String> updaterOptions = stringMapOrNull(node,
"updaterOptions");
+ return PolicyExecutionSpec.of(
+ timeTrigger, commitTrigger, minIntervalMs, jobOptions, updaterOptions);
+ }
+
+ /**
+ * Serializer for {@link PolicyExecutionSpec} as a JSON object with the flat
peer property names.
+ *
+ * <p>Same role as {@link JsonUtils.DistributionSerializer}: owns the wire
shape of the composed
+ * value. Domain content serdes call {@link #writeExecutionPeerFields} so
peers stay un-nested on
+ * the content object.
+ */
+ public static class PolicyExecutionSpecSerializer extends
JsonSerializer<PolicyExecutionSpec> {
+ @Override
+ public void serialize(
+ PolicyExecutionSpec value, JsonGenerator gen, SerializerProvider
serializers)
+ throws IOException {
+ gen.writeStartObject();
+ writeExecutionPeerFields(value, gen);
+ gen.writeEndObject();
+ }
+ }
+
+ /**
+ * Deserializer for {@link PolicyExecutionSpec} from a JSON object with the
flat peer property
+ * names.
+ */
+ public static class PolicyExecutionSpecDeserializer
+ extends JsonDeserializer<PolicyExecutionSpec> {
+ @Override
+ public PolicyExecutionSpec deserialize(JsonParser p,
DeserializationContext ctxt)
+ throws IOException {
+ JsonNode node = p.getCodec().readTree(p);
+ return readExecutionSpec(node);
+ }
+ }
+
+ /** Domain persistence serde for Iceberg compaction content. */
+ public static class IcebergDataCompactionContentSerializer
+ extends JsonSerializer<IcebergDataCompactionContent> {
+ @Override
+ public void serialize(
+ IcebergDataCompactionContent value, JsonGenerator gen,
SerializerProvider serializers)
+ throws IOException {
+ gen.writeStartObject();
+ gen.writeNumberField("minDataFileMse", value.minDataFileMse());
+ gen.writeNumberField("minDeleteFileNumber", value.minDeleteFileNumber());
+ gen.writeNumberField("dataFileMseWeight", value.dataFileMseWeight());
+ gen.writeNumberField("deleteFileNumberWeight",
value.deleteFileNumberWeight());
+ gen.writeNumberField("maxPartitionNum", value.maxPartitionNum());
+ gen.writeStringField("rewriteStrategy", value.rewriteStrategy());
+ gen.writeStringField("sortOrder", value.sortOrder());
+ if (value.rewriteOptions() != null && !value.rewriteOptions().isEmpty())
{
+ gen.writeObjectField("rewriteOptions", value.rewriteOptions());
+ }
+ writeExecutionPeerFields(value.executionSpec(), gen);
+ gen.writeEndObject();
+ }
+ }
+
+ /** Domain persistence serde for Iceberg compaction content. */
+ public static class IcebergDataCompactionContentDeserializer
+ extends JsonDeserializer<IcebergDataCompactionContent> {
+ @Override
+ public IcebergDataCompactionContent deserialize(JsonParser p,
DeserializationContext ctxt)
+ throws IOException {
+ JsonNode node = p.getCodec().readTree(p);
+ Preconditions.checkArgument(
+ node != null && node.isObject(),
+ "Cannot parse iceberg compaction content from invalid JSON: %s",
+ node);
+ return (IcebergDataCompactionContent)
+ PolicyContents.icebergDataCompaction(
+ longOrNull(node, "minDataFileMse"),
+ longOrNull(node, "minDeleteFileNumber"),
+ longOrNull(node, "dataFileMseWeight"),
+ longOrNull(node, "deleteFileNumberWeight"),
+ longOrNull(node, "maxPartitionNum"),
+ textOrNull(node, "rewriteStrategy"),
+ textOrNull(node, "sortOrder"),
+ stringMapOrNull(node, "rewriteOptions"),
+ textOrNull(node, "timeTrigger"),
+ booleanOrNull(node, "commitTrigger"),
+ longOrNull(node, "minIntervalMs"),
+ stringMapOrNull(node, "jobOptions"),
+ stringMapOrNull(node, "updaterOptions"));
+ }
+ }
+
+ /** Domain persistence serde for orphan-file removal content. */
+ public static class IcebergOrphanFileRemovalContentSerializer
+ extends JsonSerializer<IcebergOrphanFileRemovalContent> {
+ @Override
+ public void serialize(
+ IcebergOrphanFileRemovalContent value, JsonGenerator gen,
SerializerProvider serializers)
+ throws IOException {
+ gen.writeStartObject();
+ gen.writeNumberField("olderThanDays", value.olderThanDays());
+ if (value.location() != null) {
+ gen.writeStringField("location", value.location());
+ }
+ gen.writeBooleanField("dryRun", value.dryRun());
+ writeExecutionPeerFields(value.executionSpec(), gen);
+ gen.writeEndObject();
+ }
+ }
+
+ /** Domain persistence serde for orphan-file removal content. */
+ public static class IcebergOrphanFileRemovalContentDeserializer
+ extends JsonDeserializer<IcebergOrphanFileRemovalContent> {
+ @Override
+ public IcebergOrphanFileRemovalContent deserialize(JsonParser p,
DeserializationContext ctxt)
+ throws IOException {
+ JsonNode node = p.getCodec().readTree(p);
+ Preconditions.checkArgument(
+ node != null && node.isObject(),
+ "Cannot parse orphan file removal content from invalid JSON: %s",
+ node);
+ Long olderThanDaysNode = longOrNull(node, "olderThanDays");
+ long olderThanDays =
+ olderThanDaysNode == null
+ ? IcebergOrphanFileRemovalContent.DEFAULT_OLDER_THAN_DAYS
+ : olderThanDaysNode;
+ Boolean dryRunNode = booleanOrNull(node, "dryRun");
+ boolean dryRun =
+ dryRunNode == null ? IcebergOrphanFileRemovalContent.DEFAULT_DRY_RUN
: dryRunNode;
+ return (IcebergOrphanFileRemovalContent)
+ PolicyContents.icebergOrphanFileRemoval(
+ olderThanDays,
+ textOrNull(node, "location"),
+ dryRun,
+ textOrNull(node, "timeTrigger"),
+ booleanOrNull(node, "commitTrigger"),
+ longOrNull(node, "minIntervalMs"),
+ stringMapOrNull(node, "jobOptions"),
+ stringMapOrNull(node, "updaterOptions"));
+ }
+ }
+
+ /** Domain persistence serde for rewrite-manifests content. */
+ public static class IcebergRewriteManifestsContentSerializer
+ extends JsonSerializer<IcebergRewriteManifestsContent> {
+ @Override
+ public void serialize(
+ IcebergRewriteManifestsContent value, JsonGenerator gen,
SerializerProvider serializers)
+ throws IOException {
+ gen.writeStartObject();
+ writeExecutionPeerFields(value.executionSpec(), gen);
+ gen.writeEndObject();
+ }
+ }
+
+ /** Domain persistence serde for rewrite-manifests content. */
+ public static class IcebergRewriteManifestsContentDeserializer
+ extends JsonDeserializer<IcebergRewriteManifestsContent> {
+ @Override
+ public IcebergRewriteManifestsContent deserialize(JsonParser p,
DeserializationContext ctxt)
+ throws IOException {
+ JsonNode node = p.getCodec().readTree(p);
+ Preconditions.checkArgument(
+ node != null && node.isObject(),
+ "Cannot parse rewrite-manifests content from invalid JSON: %s",
+ node);
+ return (IcebergRewriteManifestsContent)
+ PolicyContents.icebergRewriteManifests(
+ textOrNull(node, "timeTrigger"),
+ booleanOrNull(node, "commitTrigger"),
+ longOrNull(node, "minIntervalMs"),
+ stringMapOrNull(node, "jobOptions"),
+ stringMapOrNull(node, "updaterOptions"));
+ }
+ }
+
+ /** Domain persistence serde for snapshot-expiration content. */
+ public static class IcebergSnapshotExpirationContentSerializer
+ extends JsonSerializer<IcebergSnapshotExpirationContent> {
+ @Override
+ public void serialize(
+ IcebergSnapshotExpirationContent value, JsonGenerator gen,
SerializerProvider serializers)
+ throws IOException {
+ gen.writeStartObject();
+ writeExecutionPeerFields(value.executionSpec(), gen);
+ gen.writeEndObject();
+ }
+ }
+
+ /** Domain persistence serde for snapshot-expiration content. */
+ public static class IcebergSnapshotExpirationContentDeserializer
+ extends JsonDeserializer<IcebergSnapshotExpirationContent> {
+ @Override
+ public IcebergSnapshotExpirationContent deserialize(JsonParser p,
DeserializationContext ctxt)
+ throws IOException {
+ JsonNode node = p.getCodec().readTree(p);
+ Preconditions.checkArgument(
+ node != null && node.isObject(),
+ "Cannot parse snapshot-expiration content from invalid JSON: %s",
+ node);
+ return (IcebergSnapshotExpirationContent)
+ PolicyContents.icebergSnapshotExpiration(
+ textOrNull(node, "timeTrigger"),
+ booleanOrNull(node, "commitTrigger"),
+ longOrNull(node, "minIntervalMs"),
+ stringMapOrNull(node, "jobOptions"),
+ stringMapOrNull(node, "updaterOptions"));
+ }
+ }
+
+ @Nullable
+ private static String textOrNull(JsonNode node, String property) {
+ if (!node.has(property) || node.get(property).isNull()) {
+ return null;
+ }
+ return node.get(property).asText();
+ }
+
+ @Nullable
+ private static Boolean booleanOrNull(JsonNode node, String property) {
+ if (!node.has(property) || node.get(property).isNull()) {
+ return null;
+ }
+ return node.get(property).asBoolean();
+ }
+
+ @Nullable
+ private static Long longOrNull(JsonNode node, String property) {
+ if (!node.has(property) || node.get(property).isNull()) {
+ return null;
+ }
+ return node.get(property).asLong();
+ }
+
+ @Nullable
+ private static Map<String, String> stringMapOrNull(JsonNode node, String
property) {
+ if (!node.has(property) || node.get(property).isNull()) {
+ return null;
+ }
+ JsonNode mapNode = node.get(property);
+ Preconditions.checkArgument(
+ mapNode.isObject(), "Cannot parse string map property %s from: %s",
property, mapNode);
+ Map<String, String> map = new LinkedHashMap<>();
+ Iterator<Map.Entry<String, JsonNode>> fields = mapNode.fields();
+ while (fields.hasNext()) {
+ Map.Entry<String, JsonNode> entry = fields.next();
+ map.put(entry.getKey(), entry.getValue().isNull() ? null :
entry.getValue().asText());
+ }
+ return map;
+ }
+}
diff --git a/common/src/main/java/org/apache/gravitino/json/JsonUtils.java
b/common/src/main/java/org/apache/gravitino/json/JsonUtils.java
index e98bc7e387..4757df5e14 100644
--- a/common/src/main/java/org/apache/gravitino/json/JsonUtils.java
+++ b/common/src/main/java/org/apache/gravitino/json/JsonUtils.java
@@ -82,6 +82,7 @@ import
org.apache.gravitino.dto.rel.partitions.IdentityPartitionDTO;
import org.apache.gravitino.dto.rel.partitions.ListPartitionDTO;
import org.apache.gravitino.dto.rel.partitions.PartitionDTO;
import org.apache.gravitino.dto.rel.partitions.RangePartitionDTO;
+import org.apache.gravitino.policy.PolicyExecutionSpec;
import org.apache.gravitino.rel.Column;
import org.apache.gravitino.rel.TableChange;
import org.apache.gravitino.rel.expressions.Expression;
@@ -312,16 +313,28 @@ public class JsonUtils {
.setVisibility(PropertyAccessor.FIELD,
JsonAutoDetect.Visibility.ANY)
.registerModule(new JavaTimeModule())
.registerModule(new Jdk8Module())
- .registerModule(
- new SimpleModule()
- .addDeserializer(Type.class, new TypeDeserializer())
- .addSerializer(Type.class, new TypeSerializer())
- .addDeserializer(Partitioning.class, new
PartitioningDeserializer())
- .addSerializer(Partitioning.class, new
PartitioningSerializer())
- .addDeserializer(Expression.class, new
ColumnDefaultValueDeserializer())
- .addSerializer(Expression.class, new
ColumnDefaultValueSerializer())
- .addDeserializer(StatisticValue.class, new
StatisticValueDeserializer())
- .addSerializer(StatisticValue.class, new
StatisticValueSerializer()));
+ .registerModule(anyFieldModule());
+ }
+
+ private static SimpleModule anyFieldModule() {
+ SimpleModule module =
+ new SimpleModule()
+ .addDeserializer(Type.class, new TypeDeserializer())
+ .addSerializer(Type.class, new TypeSerializer())
+ .addDeserializer(Partitioning.class, new
PartitioningDeserializer())
+ .addSerializer(Partitioning.class, new PartitioningSerializer())
+ .addDeserializer(Expression.class, new
ColumnDefaultValueDeserializer())
+ .addSerializer(Expression.class, new
ColumnDefaultValueSerializer())
+ .addDeserializer(StatisticValue.class, new
StatisticValueDeserializer())
+ .addSerializer(StatisticValue.class, new
StatisticValueSerializer())
+ .addSerializer(
+ PolicyExecutionSpec.class,
+ new
ExecutablePolicyContentSerde.PolicyExecutionSpecSerializer())
+ .addDeserializer(
+ PolicyExecutionSpec.class,
+ new
ExecutablePolicyContentSerde.PolicyExecutionSpecDeserializer());
+ ExecutablePolicyContentSerde.registerDomainSerdes(module);
+ return module;
}
/**
@@ -1309,6 +1322,22 @@ public class JsonUtils {
}
}
+ /**
+ * Custom JSON serializer for {@link PolicyExecutionSpec}.
+ *
+ * @see ExecutablePolicyContentSerde.PolicyExecutionSpecSerializer
+ */
+ public static class PolicyExecutionSpecSerializer
+ extends ExecutablePolicyContentSerde.PolicyExecutionSpecSerializer {}
+
+ /**
+ * Custom JSON deserializer for {@link PolicyExecutionSpec}.
+ *
+ * @see ExecutablePolicyContentSerde.PolicyExecutionSpecDeserializer
+ */
+ public static class PolicyExecutionSpecDeserializer
+ extends ExecutablePolicyContentSerde.PolicyExecutionSpecDeserializer {}
+
/** Custom JSON serializer for DistributionDTO objects. */
public static class DistributionSerializer extends
JsonSerializer<DistributionDTO> {
@Override
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/utils/TestPOConverters.java
b/core/src/test/java/org/apache/gravitino/storage/relational/utils/TestPOConverters.java
index dae145b560..e5fda37d73 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/utils/TestPOConverters.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/utils/TestPOConverters.java
@@ -21,6 +21,7 @@ package org.apache.gravitino.storage.relational.utils;
import static org.apache.gravitino.file.Fileset.LOCATION_NAME_UNKNOWN;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -67,9 +68,11 @@ import org.apache.gravitino.meta.TableStatisticEntity;
import org.apache.gravitino.meta.TagEntity;
import org.apache.gravitino.meta.TopicEntity;
import org.apache.gravitino.meta.UserEntity;
+import org.apache.gravitino.policy.IcebergDataCompactionContent;
import org.apache.gravitino.policy.Policy;
import org.apache.gravitino.policy.PolicyContent;
import org.apache.gravitino.policy.PolicyContents;
+import org.apache.gravitino.policy.PolicyExecutionSpec;
import org.apache.gravitino.rel.expressions.Expression;
import org.apache.gravitino.rel.expressions.NamedReference;
import org.apache.gravitino.rel.expressions.distributions.Distributions;
@@ -1032,6 +1035,142 @@ public class TestPOConverters {
assertEquals(expectedPolicy.content(), convertedPolicy.content());
}
+ @Test
+ public void testFromPolicyPOIcebergCompactionWithExecutionPeers() throws
JsonProcessingException {
+ PolicyContent content =
+ PolicyContents.icebergDataCompaction(
+ IcebergDataCompactionContent.DEFAULT_MIN_DATA_FILE_MSE,
+ IcebergDataCompactionContent.DEFAULT_MIN_DELETE_FILE_NUMBER,
+ IcebergDataCompactionContent.DEFAULT_DATA_FILE_MSE_WEIGHT,
+ IcebergDataCompactionContent.DEFAULT_DELETE_FILE_NUMBER_WEIGHT,
+ IcebergDataCompactionContent.DEFAULT_MAX_PARTITION_NUM,
+ IcebergDataCompactionContent.DEFAULT_REWRITE_STRATEGY,
+ IcebergDataCompactionContent.DEFAULT_SORT_ORDER,
+ ImmutableMap.of("target-file-size-bytes", "134217728"),
+ "0 0 2 * * ?",
+ true,
+ 3_600_000L,
+ ImmutableMap.of("spark.executor.memory", "4g"),
+ ImmutableMap.of("metalake", "ml"));
+
+ // Persistence must keep peer keys flat (not nested under executionSpec).
+ String storedJson = JsonUtils.anyFieldMapper().writeValueAsString(content);
+ assertTrue(storedJson.contains("\"timeTrigger\""));
+ assertTrue(storedJson.contains("\"commitTrigger\""));
+ assertTrue(storedJson.contains("\"minIntervalMs\""));
+ assertTrue(storedJson.contains("\"jobOptions\""));
+ assertTrue(storedJson.contains("\"updaterOptions\""));
+ assertFalse(storedJson.contains("\"executionSpec\""));
+
+ PolicyVersionPO policyVersionPO =
+ createPolicyVersionPO(1L, 1L, 1L, "compaction with peers", true,
content);
+ PolicyPO policyPO =
+ createPolicyPO(1L, "iceberg-compaction", "system_iceberg_compaction",
1L, policyVersionPO);
+
+ PolicyEntity convertedPolicy =
+ POConverters.fromPolicyPO(policyPO,
NamespaceUtil.ofPolicy("test_metalake"));
+
+ assertEquals(Policy.BuiltInType.ICEBERG_COMPACTION,
convertedPolicy.policyType());
+ assertEquals(content, convertedPolicy.content());
+
+ IcebergDataCompactionContent loaded = (IcebergDataCompactionContent)
convertedPolicy.content();
+ PolicyExecutionSpec peers = loaded.executionSpec();
+ assertEquals("0 0 2 * * ?", peers.timeTrigger());
+ assertEquals(Boolean.TRUE, peers.commitTrigger());
+ assertEquals(3_600_000L, peers.minIntervalMs());
+ assertEquals("4g", peers.jobOptions().get("spark.executor.memory"));
+ assertEquals("ml", peers.updaterOptions().get("metalake"));
+ assertEquals("134217728",
loaded.rewriteOptions().get("target-file-size-bytes"));
+ }
+
+ @Test
+ public void testFromPolicyPOIcebergOrphanFileRemovalWithExecutionPeers()
+ throws JsonProcessingException {
+ PolicyContent content =
+ PolicyContents.icebergOrphanFileRemoval(
+ 7L,
+ "s3://bucket/table/data",
+ true,
+ "0 0 3 * * ?",
+ null,
+ 1_800_000L,
+ ImmutableMap.of("spark.executor.cores", "2"),
+ ImmutableMap.of("metalake", "ml"));
+ assertFlatExecutionPeersPersisted(content);
+
+ PolicyEntity converted =
+ POConverters.fromPolicyPO(
+ createPolicyPO(
+ 2L,
+ "orphan-cleanup",
+ "system_iceberg_orphan_file_removal",
+ 1L,
+ createPolicyVersionPO(2L, 1L, 1L, "orphan peers", true,
content)),
+ NamespaceUtil.ofPolicy("test_metalake"));
+ assertEquals(Policy.BuiltInType.ICEBERG_ORPHAN_FILE_REMOVAL,
converted.policyType());
+ assertEquals(content, converted.content());
+ }
+
+ @Test
+ public void testFromPolicyPOIcebergRewriteManifestsWithExecutionPeers()
+ throws JsonProcessingException {
+ PolicyContent content =
+ PolicyContents.icebergRewriteManifests(
+ "0 30 1 * * ?",
+ true,
+ 900_000L,
+ ImmutableMap.of("spark.executor.memory", "2g"),
+ ImmutableMap.of("queue", "tms"));
+ assertFlatExecutionPeersPersisted(content);
+
+ PolicyEntity converted =
+ POConverters.fromPolicyPO(
+ createPolicyPO(
+ 3L,
+ "rewrite-manifests",
+ "system_iceberg_rewrite_manifests",
+ 1L,
+ createPolicyVersionPO(3L, 1L, 1L, "rewrite peers", true,
content)),
+ NamespaceUtil.ofPolicy("test_metalake"));
+ assertEquals(Policy.BuiltInType.ICEBERG_REWRITE_MANIFESTS,
converted.policyType());
+ assertEquals(content, converted.content());
+ }
+
+ @Test
+ public void testFromPolicyPOIcebergSnapshotExpirationWithExecutionPeers()
+ throws JsonProcessingException {
+ PolicyContent content =
+ PolicyContents.icebergSnapshotExpiration(
+ "0 0 4 * * ?",
+ false,
+ 2_700_000L,
+ ImmutableMap.of("spark.driver.memory", "1g"),
+ ImmutableMap.of("metalake", "ml"));
+ assertFlatExecutionPeersPersisted(content);
+
+ PolicyEntity converted =
+ POConverters.fromPolicyPO(
+ createPolicyPO(
+ 4L,
+ "snapshot-expiration",
+ "system_iceberg_snapshot_expiration",
+ 1L,
+ createPolicyVersionPO(4L, 1L, 1L, "snapshot peers", true,
content)),
+ NamespaceUtil.ofPolicy("test_metalake"));
+ assertEquals(Policy.BuiltInType.ICEBERG_SNAPSHOT_EXPIRATION,
converted.policyType());
+ assertEquals(content, converted.content());
+ }
+
+ private static void assertFlatExecutionPeersPersisted(PolicyContent content)
+ throws JsonProcessingException {
+ String storedJson = JsonUtils.anyFieldMapper().writeValueAsString(content);
+ assertTrue(storedJson.contains("\"timeTrigger\""));
+ assertTrue(storedJson.contains("\"minIntervalMs\""));
+ assertTrue(storedJson.contains("\"jobOptions\""));
+ assertTrue(storedJson.contains("\"updaterOptions\""));
+ assertFalse(storedJson.contains("\"executionSpec\""));
+ }
+
@Test
public void testFromTagPO() throws JsonProcessingException {
TagPO tagPO = createTagPO(1L, "test", 1L, "this is test");
diff --git a/design-docs/table-maintenance-service-design.md
b/design-docs/table-maintenance-service-design.md
index d4487e8e12..5ec6be113f 100644
--- a/design-docs/table-maintenance-service-design.md
+++ b/design-docs/table-maintenance-service-design.md
@@ -125,6 +125,31 @@ parts:
1. **Policy** — what to maintain, when to trigger, and non-sensitive job
settings.
2. **Scheduling and commit** — how work is enqueued, claimed, and recorded.
+### 5.0 Terminology: policy identity vs scheduler task kind
+
+Do **not** overload “task type”. TMS uses two different identities:
+
+| Concept | Meaning | Identity today |
+| ------- | ------- | -------------- |
+| **Policy / activity type** | Which maintenance the policy describes
(compaction, manifest-rewrite, snapshot-expiry, orphan-cleanup) |
`Policy.BuiltInType` (e.g. `ICEBERG_COMPACTION`) |
+| **Scheduler task kind** | Which db-scheduler row / pool owns the work |
`policy-expand`, `cron-job`, `commit-job` |
+
+Rules:
+
+- Prefer **`Policy.BuiltInType`** as the single policy/activity identity.
Server defaults
+ (for example per-type `minIntervalMs`) and content registries key off it. Do
**not**
+ introduce a parallel enum that maps 1:1 to `BuiltInType` (the former
+ `TableMaintenanceTaskType` pattern).
+- Keep **scheduler task kinds** separate. They are control-plane scheduling
units, not
+ policy types. One policy may produce expand / cron / commit work without
changing its
+ `BuiltInType`.
+- If a future runtime “activity” ever diverges from policy identity (non-1:1),
introduce
+ that concept then with an explicit contract. Until then, do not invent a
second
+ activity layer.
+
+In prose, prefer “maintenance type” / `BuiltInType` for policy activity, and
“scheduler
+task kind” for expand / cron-job / commit-job.
+
TMS has two trigger paths that stay separate:
- **crontab** — expand a policy into one-shot Spark submits (one type per
submit).
@@ -147,31 +172,37 @@ policy first, then scheduling.
TMS reuses the existing metalake Policy APIs and does **not** add
`/api/maintenance/table/policies`. Each policy covers **one** maintenance
type. The same
-policy may enable both `onCommit` and `crontab`.
+policy may enable both commit and time triggers (`commitTrigger` and
`timeTrigger`).
-#### 5.1.1 Evaluate triggers (`onCommit` + `crontab`)
+#### 5.1.1 Evaluate triggers (`commitTrigger` + `timeTrigger`)
-Triggers live in **`policy_version_info.content.schedule`**, next to
`jobOptions`,
-`rewriteOptions`, and `minIntervalMs`. Operators edit that versioned content
through Policy
-APIs. `policy_meta` only holds identity and attachment. The
`scheduled_*_tasks` tables are runtime
+Triggers are **flat** peers on policy content (same JSON object as
`jobOptions` /
+`updaterOptions` / `rewriteOptions` / `minIntervalMs`): `commitTrigger`
(boolean) and
+`timeTrigger` (cron string). Operators edit that versioned content through
Policy APIs.
+`policy_meta` only holds identity and attachment. The `scheduled_*_tasks`
tables are runtime
state for db-scheduler — not where operators set triggers.
TMS has two trigger modes. They never feed each other's queues:
-| Trigger | Path | Which types
|
-| ---------- | ---------------------------------------- |
----------------------------------------------- |
-| `onCommit` | IRC → **commit-job** only (skips expand) | Compaction /
manifest-rewrite / snapshot-expiry |
-| `crontab` | **policy-expand** → cron-job | All four types
|
+| Wire field | Runtime path | Which types
|
+| ---------------- | ----------------------------------------- |
----------------------------------------------- |
+| `commitTrigger` | IRC → **commit-job** only (skips expand) | Compaction /
manifest-rewrite / snapshot-expiry |
+| `timeTrigger` | **policy-expand** → cron-job | All four
types |
-**`onCommit`** runs after a successful Iceberg commit through IRC. That fits
types that
+**`commitTrigger`** runs after a successful Iceberg commit through IRC. That
fits types that
rewrite or expire data and metadata from the commit stream. **`orphan-cleanup`
does not:**
it scans storage for unreferenced files on a schedule. Running it on every
commit would be
-too heavy. Create/alter rejects `schedule.onCommit = true` for orphan-cleanup;
that type is
-**crontab-only**.
+too heavy. Create/alter rejects `commitTrigger = true` for orphan-cleanup;
that type is
+**timeTrigger-only** when automation is enabled.
-**`minIntervalMs`** sits in policy content **beside** `schedule` (not inside
it). It is the
-cooldown for **that policy's type** after `MAX(finished_at)` on
`table_maintenance_job` for
-`(table, policy)`. Server conf and code defaults are lower-priority fallbacks.
+**`minIntervalMs`** is the cooldown for **that policy's type** after
`MAX(finished_at)` on
+`table_maintenance_job` for `(table, policy)`. Server conf and code defaults
are
+lower-priority fallbacks.
+
+**`timeTrigger` cron:** the string is stored as-is on the policy API. Syntax
and time-zone
+rules are owned by the TMS / db-scheduler registration path (not re-validated
with a second
+parser in `api`). Examples below use a typical six-field Quartz-style form;
align examples
+with the scheduler contract when registration validation is wired.
Example content:
@@ -183,28 +214,27 @@ Example content:
"uri": "http://host:9001/iceberg";,
"type": "rest"
},
- "schedule": { "onCommit": true, "crontab": "0 2 * * *" },
+ "commitTrigger": true,
+ "timeTrigger": "0 0 2 * * ?",
"minIntervalMs": 3600000
}
```
-| Field | Runtime effect
|
-| ------------------- |
----------------------------------------------------------------- |
-| `schedule.onCommit` | IRC upserts commit-job; commit pool submits one
chained Job |
-| `schedule.crontab` | Sets / refreshes next `policy-expand` due time
|
-| `minIntervalMs` | Cooldown for this policy's type; highest when set
|
-| `jobOptions` | Non-sensitive Spark / IRC client props (`uri`, `type`,
memory, …) |
-
-Automated maintenance should set at least one of `onCommit` or `crontab`
(orphan-cleanup:
-**`crontab` only**).
+| Field | Runtime effect
|
+| ---------------- |
----------------------------------------------------------------- |
+| `commitTrigger` | IRC upserts commit-job; commit pool submits one chained
Job |
+| `timeTrigger` | Sets / refreshes next `policy-expand` due time
|
+| `minIntervalMs` | Cooldown for this policy's type; highest when set
|
+| `jobOptions` | Non-sensitive Spark / IRC client props (`uri`, `type`,
memory, …) |
+| `updaterOptions` | Non-sensitive Gravitino updater props (`gravitino_uri`,
`metalake`, …) |
#### 5.1.2 Sensitive credentials
Policy content is versioned, edited through Policy APIs, and readable with
`VIEW_POLICY`.
**Secrets must not live there.** Put credential-shaped keys in `jobOptions` /
-`rewriteOptions` and create/alter **rejects** them. Non-sensitive Spark / IRC
client props
-(`uri`, `type`, memory, …) stay on the nearest policy; auth is cluster-static
in
-`gravitino.conf` (values may be SecretManager references). Conf keys: §7.4.
+`rewriteOptions` / `updaterOptions` and create/alter **rejects** them.
Non-sensitive Spark /
+IRC client props (`uri`, `type`, memory, …) stay on the nearest policy; auth is
+cluster-static in `gravitino.conf` (values may be SecretManager references).
Conf keys: §7.4.
TMS talks to **two** backends with **two** identities, so there are two
prefixes:
@@ -220,7 +250,7 @@ each — it does not write secrets back into policy content:
| Runtime bag | Non-sensitive (examples)
| Auth overlay |
| ------------------- |
------------------------------------------------------------- |
--------------------------------------- |
-| **`updateOptions`** | `gravitino_uri`, `metalake`, updater impl names, …
| `gravitino.maintenance.gravitinoAuth.*` |
+| **`updaterOptions`** | `gravitino_uri`, `metalake`, updater impl names, …
| `gravitino.maintenance.gravitinoAuth.*` |
| **`jobOptions`** | nearest policy `jobOptions` (`uri`, `type`, Spark
resources…) | `gravitino.maintenance.ircAuth.*` |
### 5.2 Scheduling and commit
@@ -325,7 +355,7 @@ flowchart TD
B --> C[Keep expand row; set next crontab]
```
-Example with crontab `0 2 * * *`: create inserts expand due next 02:00; at
02:00 expand inserts
+Example with `timeTrigger` `0 0 2 * * ?`: create inserts expand due next
02:00; at 02:00 expand inserts
many cron-jobs due now, then schedules itself for the next 02:00. Expand's due
time is when
**enqueue** happens — not when Spark runs.
@@ -500,7 +530,7 @@ Cluster-static — **not** in policy content and **not** a DB
table.
| Prefix | Role
| Merged into `runJob` bag |
| -------------------------------------- |
------------------------------------------------------------ |
----------------------------------------------- |
-| `gravitino.maintenance.gravitinoAuth.` | Expand / submit / RBAC client; also
jobs that call Gravitino | `updateOptions` (+ non-sensitive updater props) |
+| `gravitino.maintenance.gravitinoAuth.` | Expand / submit / RBAC client; also
jobs that call Gravitino | `updaterOptions` (+ non-sensitive updater props) |
| `gravitino.maintenance.ircAuth.` | Iceberg REST client auth for Spark
Jobs → IRC | `jobOptions` (+ policy non-sensitive props) |
Illustrative auth keys (same shape under each prefix; values may be
SecretManager references where
