@@ -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());
+  }
+}
@@ -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();
+}
@@ -32,7 +32,7 @@
 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 @@ private IcebergDataCompactionContent() {
       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 Map<String, Object> rules() {
 
   @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 boolean equals(Object o) {
       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 boolean equals(Object o) {
   @Override
   public int hashCode() {
     return Objects.hash(
+        executionFieldsHashCode(),
         minDataFileMse,
         minDeleteFileNumber,
         dataFileMseWeight,
@@ -339,25 +364,30 @@ public int hashCode() {
 
   @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();
   }
 }
@@ -31,7 +31,7 @@
 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 static void validateOlderThanDays(long days) {
 
   @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 boolean equals(Object other) {
       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);
   }
 }
@@ -24,15 +24,16 @@
 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 Map<String, Object> rules() {
 
   @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();
   }
 }