This is an automated email from the ASF dual-hosted git repository. roryqi pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/gravitino.git
commit b27fe4d9a4643895a50c041260291538ae25e6bf Author: MaSai <[email protected]> AuthorDate: Mon Oct 26 16:50:00 2026 +0800 [#14227] refactor(policy): Align PolicyExecutionSpec with module value-object conventions (#14228) Align `PolicyExecutionSpec` with existing policy-module value objects (`TagValueSelector` / content factories): - Drop `withXxx` copy-mutators; construct only via `of` / `empty` - Enforce shared peer constraints in `of` (no deferred `validate()`) - Store option maps as `ImmutableMap`; add `@Evolving` - DTO peer setters rebuild via `ExecutionPeersDTO` + `PolicyExecutionSpec.of` - Update unit tests for construction-time validation Keep the TMS execution-peer value type consistent with the rest of the `api` policy package. Fix: #14227 Yes — breaking for callers of removed `PolicyExecutionSpec.with*` / `validate()`. Wire JSON peers unchanged. ```bash ./gradlew :api:test --tests 'org.apache.gravitino.policy.TestPolicy*' \ :common:test --tests 'org.apache.gravitino.dto.policy.TestPolicyDTO' \ --tests 'org.apache.gravitino.dto.policy.TestOrphanFileRemovalPolicyDTO' \ --tests 'org.apache.gravitino.dto.policy.TestRewriteManifestsAndSnapshotExpirationPolicyDTO' \ -PskipITs ``` Made with [Cursor](https://cursor.com) --------- Co-authored-by: Cursor <[email protected]> --- .../policy/AbstractExecutablePolicyContent.java | 8 +- .../gravitino/policy/PolicyExecutionSpec.java | 117 ++---- .../gravitino/policy/TestPolicyContents.java | 42 +- .../gravitino/policy/TestPolicyExecutionSpec.java | 53 ++- .../gravitino/dto/policy/PolicyContentDTO.java | 460 +++++++++++++-------- 5 files changed, 397 insertions(+), 283 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 index c24a827699..1b6d1ccced 100644 --- a/api/src/main/java/org/apache/gravitino/policy/AbstractExecutablePolicyContent.java +++ b/api/src/main/java/org/apache/gravitino/policy/AbstractExecutablePolicyContent.java @@ -55,16 +55,16 @@ public abstract class AbstractExecutablePolicyContent implements ExecutablePolic } /** - * Validates universal {@link PolicyContent} rules plus shared execution-peer checks. + * Validates universal {@link PolicyContent} rules. * - * <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 + * <p>Shared execution-peer constraints are enforced when constructing {@link + * PolicyExecutionSpec}. Type-specific maintenance contracts (for example orphan cleanup + * forbidding {@code commitTrigger}, or cron parsing) 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(); } /** diff --git a/api/src/main/java/org/apache/gravitino/policy/PolicyExecutionSpec.java b/api/src/main/java/org/apache/gravitino/policy/PolicyExecutionSpec.java index 24ab7ba572..9783c65581 100644 --- a/api/src/main/java/org/apache/gravitino/policy/PolicyExecutionSpec.java +++ b/api/src/main/java/org/apache/gravitino/policy/PolicyExecutionSpec.java @@ -19,12 +19,12 @@ package org.apache.gravitino.policy; import com.google.common.base.Preconditions; -import java.util.Collections; -import java.util.LinkedHashMap; +import com.google.common.collect.ImmutableMap; import java.util.Map; import java.util.Objects; import javax.annotation.Nullable; import org.apache.commons.lang3.StringUtils; +import org.apache.gravitino.annotation.Evolving; /** * Immutable execution peer fields for executable policy contents: triggers, cooldown, and option @@ -36,20 +36,21 @@ import org.apache.commons.lang3.StringUtils; * 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>Shared peer constraints are checked in {@link #of}; 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. */ +@Evolving public final class PolicyExecutionSpec { /** Spec with every peer unset. */ public static final PolicyExecutionSpec EMPTY = - new PolicyExecutionSpec(null, null, null, null, null); + new PolicyExecutionSpec(null, null, null, ImmutableMap.of(), ImmutableMap.of()); @Nullable private final String timeTrigger; @Nullable private final Boolean commitTrigger; @@ -61,13 +62,13 @@ public final class 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; + Map<String, String> jobOptions, + Map<String, String> updaterOptions) { + this.timeTrigger = timeTrigger; this.commitTrigger = commitTrigger; this.minIntervalMs = minIntervalMs; - this.jobOptions = copyOptions(jobOptions); - this.updaterOptions = copyOptions(updaterOptions); + this.jobOptions = jobOptions; + this.updaterOptions = updaterOptions; } /** @@ -80,6 +81,8 @@ public final class PolicyExecutionSpec { * @param jobOptions optional job options * @param updaterOptions optional updater options * @return immutable spec, or {@link #EMPTY} when every peer is unset + * @throws IllegalArgumentException if {@code minIntervalMs} is negative or an option map has a + * blank key or value */ public static PolicyExecutionSpec of( @Nullable String timeTrigger, @@ -87,11 +90,28 @@ public final class PolicyExecutionSpec { @Nullable Long minIntervalMs, @Nullable Map<String, String> jobOptions, @Nullable Map<String, String> updaterOptions) { - if (!hasAny(timeTrigger, commitTrigger, minIntervalMs, jobOptions, updaterOptions)) { + String normalizedTimeTrigger = StringUtils.isBlank(timeTrigger) ? null : timeTrigger; + Map<String, String> normalizedJobOptions = copyOptions(jobOptions); + Map<String, String> normalizedUpdaterOptions = copyOptions(updaterOptions); + if (minIntervalMs != null) { + Preconditions.checkArgument(minIntervalMs >= 0, "minIntervalMs must be >= 0"); + } + validateOptionMap(normalizedJobOptions, "jobOptions"); + validateOptionMap(normalizedUpdaterOptions, "updaterOptions"); + if (!hasAny( + normalizedTimeTrigger, + commitTrigger, + minIntervalMs, + normalizedJobOptions, + normalizedUpdaterOptions)) { return EMPTY; } return new PolicyExecutionSpec( - timeTrigger, commitTrigger, minIntervalMs, jobOptions, updaterOptions); + normalizedTimeTrigger, + commitTrigger, + minIntervalMs, + normalizedJobOptions, + normalizedUpdaterOptions); } /** @@ -147,63 +167,6 @@ public final class PolicyExecutionSpec { 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) { @@ -253,24 +216,24 @@ public final class PolicyExecutionSpec { @Nullable String timeTrigger, @Nullable Boolean commitTrigger, @Nullable Long minIntervalMs, - @Nullable Map<String, String> jobOptions, - @Nullable Map<String, String> updaterOptions) { + Map<String, String> jobOptions, + Map<String, String> updaterOptions) { return StringUtils.isNotBlank(timeTrigger) || commitTrigger != null || minIntervalMs != null - || (jobOptions != null && !jobOptions.isEmpty()) - || (updaterOptions != null && !updaterOptions.isEmpty()); + || !jobOptions.isEmpty() + || !updaterOptions.isEmpty(); } private static Map<String, String> copyOptions(@Nullable Map<String, String> options) { if (options == null || options.isEmpty()) { - return Collections.emptyMap(); + return ImmutableMap.of(); } - return Collections.unmodifiableMap(new LinkedHashMap<>(options)); + return ImmutableMap.copyOf(options); } private static void validateOptionMap(Map<String, String> options, String mapName) { - if (options == null || options.isEmpty()) { + if (options.isEmpty()) { return; } options.forEach( 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 b03541c567..bcd8e3a780 100644 --- a/api/src/test/java/org/apache/gravitino/policy/TestPolicyContents.java +++ b/api/src/test/java/org/apache/gravitino/policy/TestPolicyContents.java @@ -177,26 +177,29 @@ public class TestPolicyContents { IllegalArgumentException exception = Assertions.assertThrows(IllegalArgumentException.class, content::validate); - Assertions.assertTrue(exception.g @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); + Assertions.assertTrue(exception.getMessage().contains("maxPartitionNum")); + } + @Test + void testOfRejectsNegativeMinIntervalMsOnExecutableContent() { IllegalArgumentException exception = - Assertions.assertThrows(IllegalArgumentException.class, content::validate); + Assertions.assertThrows( + IllegalArgumentException.class, + () -> + 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)); Assertions.assertTrue(exception.getMessage().contains("minIntervalMs")); } @@ -241,9 +244,6 @@ public class TestPolicyContents { Assertions.assertTrue(exception.getMessage().contains("commitTrigger")); } -etMessage().contains("maxPartitionNum")); - } - @Test void testIcebergCompactionContentRejectsUnknownOrInvalidRewriteOptions() { IllegalArgumentException unknown = diff --git a/api/src/test/java/org/apache/gravitino/policy/TestPolicyExecutionSpec.java b/api/src/test/java/org/apache/gravitino/policy/TestPolicyExecutionSpec.java index cf36a06d66..07bff89f56 100644 --- a/api/src/test/java/org/apache/gravitino/policy/TestPolicyExecutionSpec.java +++ b/api/src/test/java/org/apache/gravitino/policy/TestPolicyExecutionSpec.java @@ -58,31 +58,35 @@ public class TestPolicyExecutionSpec { } @Test - void validateAcceptsScalarTriggers() { + void ofAcceptsScalarTriggers() { PolicyExecutionSpec spec = - PolicyExecutionSpec.of( - "0 0 2 * * ?", - true, - 100L, - ImmutableMap.of("uri", "http://irc";), - ImmutableMap.of("metalake", "ml")); - Assertions.assertDoesNotThrow(spec::validate); + Assertions.assertDoesNotThrow( + () -> + PolicyExecutionSpec.of( + "0 0 2 * * ?", + true, + 100L, + ImmutableMap.of("uri", "http://irc";), + ImmutableMap.of("metalake", "ml"))); + Assertions.assertEquals("0 0 2 * * ?", spec.timeTrigger()); + Assertions.assertEquals(Boolean.TRUE, spec.commitTrigger()); } @Test - void validateAcceptsCommitTriggerTrueAlone() { - PolicyExecutionSpec spec = - PolicyExecutionSpec.of(null, true, null, Collections.emptyMap(), Collections.emptyMap()); - Assertions.assertDoesNotThrow(spec::validate); + void ofAcceptsCommitTriggerTrueAlone() { + Assertions.assertDoesNotThrow( + () -> + PolicyExecutionSpec.of( + null, true, null, Collections.emptyMap(), Collections.emptyMap())); } @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); + Assertions.assertDoesNotThrow( + () -> + PolicyExecutionSpec.of( + "0 2 * *", null, null, Collections.emptyMap(), Collections.emptyMap())); } @Test @@ -90,14 +94,23 @@ public class TestPolicyExecutionSpec { 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); + void ofRejectsNegativeMinInterval() { IllegalArgumentException ex = - Assertions.assertThrows(IllegalArgumentException.class, spec::validate); + Assertions.assertThrows( + IllegalArgumentException.class, + () -> PolicyExecutionSpec.of(null, null, -1L, null, null)); Assertions.assertTrue(ex.getMessage().contains("minIntervalMs")); } + + @Test + void ofRejectsBlankOptionValue() { + IllegalArgumentException ex = + Assertions.assertThrows( + IllegalArgumentException.class, + () -> PolicyExecutionSpec.of(null, null, null, ImmutableMap.of("uri", " "), null)); + Assertions.assertTrue(ex.getMessage().contains("jobOptions")); + } } diff --git a/common/src/main/java/org/apache/gravitino/dto/policy/PolicyContentDTO.java b/common/src/main/java/org/apache/gravitino/dto/policy/PolicyContentDTO.java index 990c51cf29..7e10253286 100644 --- a/common/src/main/java/org/apache/gravitino/dto/policy/PolicyContentDTO.java +++ b/common/src/main/java/org/apache/gravitino/dto/policy/PolicyContentDTO.java @@ -58,88 +58,63 @@ public interface PolicyContentDTO extends PolicyContent { /** * Optional maintenance execution peers on Iceberg TMS policy content DTOs. * - * <p>Peers are stored as a composed {@link PolicyExecutionSpec}. Flat JSON keys are exposed via - * {@code @JsonProperty} accessors so the wire contract stays flat without duplicating five fields - * on every DTO. + * <p>Implementations store flat peer fields (same JSON keys as the wire contract). {@link + * #executionSpec()} assembles them into a {@link PolicyExecutionSpec} at the DTO/domain boundary. * - * <p>Option maps always return unmodifiable defensive copies so callers cannot mutate DTO state - * after validation or change Lombok equality/hash state through the returned map. + * <p>{@link #jobOptions()} / {@link #updaterOptions()} return unmodifiable defensive copies so + * callers cannot mutate DTO state after validation or change Lombok equality through the returned + * map. */ interface ExecutionPeersDTO { /** - * Stores {@code null} when every peer is unset so Lombok equality matches Jackson round-trips. - * - * @param spec peers to store; empty specs are normalized to {@code null} - * @return {@code null} when unset, otherwise the spec + * @return optional cron expression, or {@code null} when unset */ - static PolicyExecutionSpec normalize(@Nullable PolicyExecutionSpec spec) { - return spec == null || !spec.hasAny() ? null : spec; - } + @Nullable + String timeTrigger(); /** - * @return immutable execution peers; never {@code null} + * @return {@code true}/{@code false} when explicitly set, or {@code null} when unset */ - @JsonIgnore - PolicyExecutionSpec executionSpec(); + @Nullable + Boolean commitTrigger(); /** - * Optional cron expression for automated evaluation. - * - * @return cron string, or {@code null} when unset + * @return cooldown milliseconds, or {@code null} when unset */ @Nullable - @JsonProperty("timeTrigger") - @JsonInclude(JsonInclude.Include.NON_NULL) - default String timeTrigger() { - return executionSpec().timeTrigger(); - } + Long minIntervalMs(); /** - * Optional flag to evaluate the policy after an Iceberg commit. - * - * @return {@code true} when enabled, {@code false} when explicitly disabled, or {@code null} - * when unset + * @return unmodifiable defensive copy of job options; never {@code null} */ - @Nullable - @JsonProperty("commitTrigger") - @JsonInclude(JsonInclude.Include.NON_NULL) - default Boolean commitTrigger() { - return executionSpec().commitTrigger(); - } + Map<String, String> jobOptions(); /** - * Optional cooldown between finished automated runs. - * - * @return milliseconds, or {@code null} when unset + * @return unmodifiable defensive copy of updater options; never {@code null} */ - @Nullable - @JsonProperty("minIntervalMs") - @JsonInclude(JsonInclude.Include.NON_NULL) - default Long minIntervalMs() { - return executionSpec().minIntervalMs(); - } + Map<String, String> updaterOptions(); /** - * Optional non-secret job / client options. - * - * @return unmodifiable defensive copy; never {@code null} (empty when unset) + * @return immutable execution peers assembled from flat fields; never {@code null} */ - @JsonProperty("jobOptions") - @JsonInclude(JsonInclude.Include.NON_EMPTY) - default Map<String, String> jobOptions() { - return executionSpec().jobOptions(); + @JsonIgnore + default PolicyExecutionSpec executionSpec() { + return PolicyExecutionSpec.of( + timeTrigger(), commitTrigger(), minIntervalMs(), jobOptions(), updaterOptions()); } /** - * Optional non-secret updater / control-plane options. + * Defensive copy for option-map accessors. * - * @return unmodifiable defensive copy; never {@code null} (empty when unset) + * @param options raw field value + * @return unmodifiable copy; empty when {@code null} or empty */ - @JsonProperty("updaterOptions") - @JsonInclude(JsonInclude.Include.NON_EMPTY) - default Map<String, String> updaterOptions() { - return executionSpec().updaterOptions(); + static Map<String, String> unmodifiableOptions(@Nullable Map<String, String> options) { + if (options == null || options.isEmpty()) { + return Collections.emptyMap(); + } + return Collections.unmodifiableMap(new LinkedHashMap<>(options)); } } @@ -189,7 +164,7 @@ public interface PolicyContentDTO extends PolicyContent { /** Represents a typed iceberg compaction policy content DTO. */ @EqualsAndHashCode - @ToString(exclude = {"executionSpec", "rewriteOptions"}) + @ToString(exclude = {"rewriteOptions", "jobOptions", "updaterOptions"}) @Builder(setterPrefix = "with") @AllArgsConstructor(access = lombok.AccessLevel.PRIVATE) class IcebergCompactionContentDTO implements PolicyContentDTO, ExecutionPeersDTO { @@ -218,12 +193,33 @@ public interface PolicyContentDTO extends PolicyContent { @JsonProperty("rewriteOptions") private Map<String, String> rewriteOptions; - @EqualsAndHashCode.Exclude @JsonIgnore @Nullable private PolicyExecutionSpec executionSpec; + @EqualsAndHashCode.Exclude + @Nullable + @JsonProperty("timeTrigger") + @JsonInclude(JsonInclude.Include.NON_NULL) + private String timeTrigger; - @EqualsAndHashCode.Include - private PolicyExecutionSpec executionSpecForEquals() { - return ExecutionPeersDTO.normalize(executionSpec); - } + @EqualsAndHashCode.Exclude + @Nullable + @JsonProperty("commitTrigger") + @JsonInclude(JsonInclude.Include.NON_NULL) + private Boolean commitTrigger; + + @EqualsAndHashCode.Exclude + @Nullable + @JsonProperty("minIntervalMs") + @JsonInclude(JsonInclude.Include.NON_NULL) + private Long minIntervalMs; + + @EqualsAndHashCode.Exclude + @JsonProperty("jobOptions") + @JsonInclude(JsonInclude.Include.NON_EMPTY) + private Map<String, String> jobOptions; + + @EqualsAndHashCode.Exclude + @JsonProperty("updaterOptions") + @JsonInclude(JsonInclude.Include.NON_EMPTY) + private Map<String, String> updaterOptions; // Default constructor for Jackson deserialization only. private IcebergCompactionContentDTO() {} @@ -315,38 +311,37 @@ public interface PolicyContentDTO extends PolicyContent { } @Override - @JsonIgnore - public PolicyExecutionSpec executionSpec() { - return executionSpec == null ? PolicyExecutionSpec.empty() : executionSpec; + @Nullable + public String timeTrigger() { + return timeTrigger; } - @JsonProperty("timeTrigger") - private void setTimeTrigger(String timeTrigger) { - this.executionSpec = - ExecutionPeersDTO.normalize(executionSpec().withTimeTrigger(timeTrigger)); + @Override + @Nullable + public Boolean commitTrigger() { + return commitTrigger; } - @JsonProperty("commitTrigger") - private void setCommitTrigger(Boolean commitTrigger) { - this.executionSpec = - ExecutionPeersDTO.normalize(executionSpec().withCommitTrigger(commitTrigger)); + @Override + @Nullable + public Long minIntervalMs() { + return minIntervalMs; } - @JsonProperty("minIntervalMs") - private void setMinIntervalMs(Long minIntervalMs) { - this.executionSpec = - ExecutionPeersDTO.normalize(executionSpec().withMinIntervalMs(minIntervalMs)); + @Override + public Map<String, String> jobOptions() { + return ExecutionPeersDTO.unmodifiableOptions(jobOptions); } - @JsonProperty("jobOptions") - private void setJobOptions(Map<String, String> jobOptions) { - this.executionSpec = ExecutionPeersDTO.normalize(executionSpec().withJobOptions(jobOptions)); + @Override + public Map<String, String> updaterOptions() { + return ExecutionPeersDTO.unmodifiableOptions(updaterOptions); } - @JsonProperty("updaterOptions") - private void setUpdaterOptions(Map<String, String> updaterOptions) { - this.executionSpec = - ExecutionPeersDTO.normalize(executionSpec().withUpdaterOptions(updaterOptions)); + @EqualsAndHashCode.Include + private PolicyExecutionSpec executionSpecForEquals() { + return PolicyExecutionSpec.of( + timeTrigger, commitTrigger, minIntervalMs, jobOptions, updaterOptions); } @Override @@ -390,11 +385,37 @@ public interface PolicyContentDTO extends PolicyContent { spec.jobOptions(), spec.updaterOptions()); } + + /** Lombok builder customization: unpack PolicyExecutionSpec into flat peer fields. */ + public static class IcebergCompactionContentDTOBuilder { + /** + * @param executionSpec peers to copy; {@code null} or empty leaves peer fields unset + * @return this builder + */ + public IcebergCompactionContentDTOBuilder withExecutionSpec( + @Nullable PolicyExecutionSpec executionSpec) { + if (executionSpec == null || !executionSpec.hasAny()) { + return this; + } + this.timeTrigger = executionSpec.timeTrigger(); + this.commitTrigger = executionSpec.commitTrigger(); + this.minIntervalMs = executionSpec.minIntervalMs(); + this.jobOptions = + executionSpec.jobOptions().isEmpty() + ? null + : new LinkedHashMap<>(executionSpec.jobOptions()); + this.updaterOptions = + executionSpec.updaterOptions().isEmpty() + ? null + : new LinkedHashMap<>(executionSpec.updaterOptions()); + return this; + } + } } /** Typed orphan cleanup policy content for REST requests and responses. */ @EqualsAndHashCode - @ToString(exclude = {"executionSpec"}) + @ToString(exclude = {"jobOptions", "updaterOptions"}) @Builder(setterPrefix = "with") @AllArgsConstructor(access = lombok.AccessLevel.PRIVATE) class IcebergOrphanFileRemovalContentDTO implements PolicyContentDTO, ExecutionPeersDTO { @@ -407,12 +428,33 @@ public interface PolicyContentDTO extends PolicyContent { @JsonProperty("dryRun") private Boolean dryRun; - @EqualsAndHashCode.Exclude @JsonIgnore @Nullable private PolicyExecutionSpec executionSpec; + @EqualsAndHashCode.Exclude + @Nullable + @JsonProperty("timeTrigger") + @JsonInclude(JsonInclude.Include.NON_NULL) + private String timeTrigger; - @EqualsAndHashCode.Include - private PolicyExecutionSpec executionSpecForEquals() { - return ExecutionPeersDTO.normalize(executionSpec); - } + @EqualsAndHashCode.Exclude + @Nullable + @JsonProperty("commitTrigger") + @JsonInclude(JsonInclude.Include.NON_NULL) + private Boolean commitTrigger; + + @EqualsAndHashCode.Exclude + @Nullable + @JsonProperty("minIntervalMs") + @JsonInclude(JsonInclude.Include.NON_NULL) + private Long minIntervalMs; + + @EqualsAndHashCode.Exclude + @JsonProperty("jobOptions") + @JsonInclude(JsonInclude.Include.NON_EMPTY) + private Map<String, String> jobOptions; + + @EqualsAndHashCode.Exclude + @JsonProperty("updaterOptions") + @JsonInclude(JsonInclude.Include.NON_EMPTY) + private Map<String, String> updaterOptions; private IcebergOrphanFileRemovalContentDTO() {} @@ -441,38 +483,37 @@ public interface PolicyContentDTO extends PolicyContent { } @Override - @JsonIgnore - public PolicyExecutionSpec executionSpec() { - return executionSpec == null ? PolicyExecutionSpec.empty() : executionSpec; + @Nullable + public String timeTrigger() { + return timeTrigger; } - @JsonProperty("timeTrigger") - private void setTimeTrigger(String timeTrigger) { - this.executionSpec = - ExecutionPeersDTO.normalize(executionSpec().withTimeTrigger(timeTrigger)); + @Override + @Nullable + public Boolean commitTrigger() { + return commitTrigger; } - @JsonProperty("commitTrigger") - private void setCommitTrigger(Boolean commitTrigger) { - this.executionSpec = - ExecutionPeersDTO.normalize(executionSpec().withCommitTrigger(commitTrigger)); + @Override + @Nullable + public Long minIntervalMs() { + return minIntervalMs; } - @JsonProperty("minIntervalMs") - private void setMinIntervalMs(Long minIntervalMs) { - this.executionSpec = - ExecutionPeersDTO.normalize(executionSpec().withMinIntervalMs(minIntervalMs)); + @Override + public Map<String, String> jobOptions() { + return ExecutionPeersDTO.unmodifiableOptions(jobOptions); } - @JsonProperty("jobOptions") - private void setJobOptions(Map<String, String> jobOptions) { - this.executionSpec = ExecutionPeersDTO.normalize(executionSpec().withJobOptions(jobOptions)); + @Override + public Map<String, String> updaterOptions() { + return ExecutionPeersDTO.unmodifiableOptions(updaterOptions); } - @JsonProperty("updaterOptions") - private void setUpdaterOptions(Map<String, String> updaterOptions) { - this.executionSpec = - ExecutionPeersDTO.normalize(executionSpec().withUpdaterOptions(updaterOptions)); + @EqualsAndHashCode.Include + private PolicyExecutionSpec executionSpecForEquals() { + return PolicyExecutionSpec.of( + timeTrigger, commitTrigger, minIntervalMs, jobOptions, updaterOptions); } @Override @@ -514,11 +555,37 @@ public interface PolicyContentDTO extends PolicyContent { spec.jobOptions(), spec.updaterOptions()); } + + /** Lombok builder customization: unpack PolicyExecutionSpec into flat peer fields. */ + public static class IcebergOrphanFileRemovalContentDTOBuilder { + /** + * @param executionSpec peers to copy; {@code null} or empty leaves peer fields unset + * @return this builder + */ + public IcebergOrphanFileRemovalContentDTOBuilder withExecutionSpec( + @Nullable PolicyExecutionSpec executionSpec) { + if (executionSpec == null || !executionSpec.hasAny()) { + return this; + } + this.timeTrigger = executionSpec.timeTrigger(); + this.commitTrigger = executionSpec.commitTrigger(); + this.minIntervalMs = executionSpec.minIntervalMs(); + this.jobOptions = + executionSpec.jobOptions().isEmpty() + ? null + : new LinkedHashMap<>(executionSpec.jobOptions()); + this.updaterOptions = + executionSpec.updaterOptions().isEmpty() + ? null + : new LinkedHashMap<>(executionSpec.updaterOptions()); + return this; + } + } } /** Typed manifest rewrite policy content. */ @EqualsAndHashCode - @ToString(exclude = {"executionSpec"}) + @ToString(exclude = {"jobOptions", "updaterOptions"}) @Builder(setterPrefix = "with") @AllArgsConstructor(access = AccessLevel.PRIVATE) class IcebergRewriteManifestsContentDTO implements PolicyContentDTO, ExecutionPeersDTO { @@ -542,12 +609,33 @@ public interface PolicyContentDTO extends PolicyContent { @JsonDeserialize(using = ManifestUseCachingDeserializer.class) private Boolean useCaching; - @EqualsAndHashCode.Exclude @JsonIgnore @Nullable private PolicyExecutionSpec executionSpec; + @EqualsAndHashCode.Exclude + @Nullable + @JsonProperty("timeTrigger") + @JsonInclude(JsonInclude.Include.NON_NULL) + private String timeTrigger; - @EqualsAndHashCode.Include - private PolicyExecutionSpec executionSpecForEquals() { - return ExecutionPeersDTO.normalize(executionSpec); - } + @EqualsAndHashCode.Exclude + @Nullable + @JsonProperty("commitTrigger") + @JsonInclude(JsonInclude.Include.NON_NULL) + private Boolean commitTrigger; + + @EqualsAndHashCode.Exclude + @Nullable + @JsonProperty("minIntervalMs") + @JsonInclude(JsonInclude.Include.NON_NULL) + private Long minIntervalMs; + + @EqualsAndHashCode.Exclude + @JsonProperty("jobOptions") + @JsonInclude(JsonInclude.Include.NON_EMPTY) + private Map<String, String> jobOptions; + + @EqualsAndHashCode.Exclude + @JsonProperty("updaterOptions") + @JsonInclude(JsonInclude.Include.NON_EMPTY) + private Map<String, String> updaterOptions; private IcebergRewriteManifestsContentDTO() {} @@ -614,38 +702,37 @@ public interface PolicyContentDTO extends PolicyContent { } @Override - @JsonIgnore - public PolicyExecutionSpec executionSpec() { - return executionSpec == null ? PolicyExecutionSpec.empty() : executionSpec; + @Nullable + public String timeTrigger() { + return timeTrigger; } - @JsonProperty("timeTrigger") - private void setTimeTrigger(String timeTrigger) { - this.executionSpec = - ExecutionPeersDTO.normalize(executionSpec().withTimeTrigger(timeTrigger)); + @Override + @Nullable + public Boolean commitTrigger() { + return commitTrigger; } - @JsonProperty("commitTrigger") - private void setCommitTrigger(Boolean commitTrigger) { - this.executionSpec = - ExecutionPeersDTO.normalize(executionSpec().withCommitTrigger(commitTrigger)); + @Override + @Nullable + public Long minIntervalMs() { + return minIntervalMs; } - @JsonProperty("minIntervalMs") - private void setMinIntervalMs(Long minIntervalMs) { - this.executionSpec = - ExecutionPeersDTO.normalize(executionSpec().withMinIntervalMs(minIntervalMs)); + @Override + public Map<String, String> jobOptions() { + return ExecutionPeersDTO.unmodifiableOptions(jobOptions); } - @JsonProperty("jobOptions") - private void setJobOptions(Map<String, String> jobOptions) { - this.executionSpec = ExecutionPeersDTO.normalize(executionSpec().withJobOptions(jobOptions)); + @Override + public Map<String, String> updaterOptions() { + return ExecutionPeersDTO.unmodifiableOptions(updaterOptions); } - @JsonProperty("updaterOptions") - private void setUpdaterOptions(Map<String, String> updaterOptions) { - this.executionSpec = - ExecutionPeersDTO.normalize(executionSpec().withUpdaterOptions(updaterOptions)); + @EqualsAndHashCode.Include + private PolicyExecutionSpec executionSpecForEquals() { + return PolicyExecutionSpec.of( + timeTrigger, commitTrigger, minIntervalMs, jobOptions, updaterOptions); } @Override @@ -723,6 +810,32 @@ public interface PolicyContentDTO extends PolicyContent { spec.jobOptions(), spec.updaterOptions()); } + + /** Lombok builder customization: unpack PolicyExecutionSpec into flat peer fields. */ + public static class IcebergRewriteManifestsContentDTOBuilder { + /** + * @param executionSpec peers to copy; {@code null} or empty leaves peer fields unset + * @return this builder + */ + public IcebergRewriteManifestsContentDTOBuilder withExecutionSpec( + @Nullable PolicyExecutionSpec executionSpec) { + if (executionSpec == null || !executionSpec.hasAny()) { + return this; + } + this.timeTrigger = executionSpec.timeTrigger(); + this.commitTrigger = executionSpec.commitTrigger(); + this.minIntervalMs = executionSpec.minIntervalMs(); + this.jobOptions = + executionSpec.jobOptions().isEmpty() + ? null + : new LinkedHashMap<>(executionSpec.jobOptions()); + this.updaterOptions = + executionSpec.updaterOptions().isEmpty() + ? null + : new LinkedHashMap<>(executionSpec.updaterOptions()); + return this; + } + } } /** @@ -776,38 +889,37 @@ public interface PolicyContentDTO extends PolicyContent { } @Override - @JsonIgnore - public PolicyExecutionSpec executionSpec() { - return executionSpec == null ? PolicyExecutionSpec.empty() : executionSpec; + @Nullable + public String timeTrigger() { + return timeTrigger; } - @JsonProperty("timeTrigger") - private void setTimeTrigger(String timeTrigger) { - this.executionSpec = - ExecutionPeersDTO.normalize(executionSpec().withTimeTrigger(timeTrigger)); + @Override + @Nullable + public Boolean commitTrigger() { + return commitTrigger; } - @JsonProperty("commitTrigger") - private void setCommitTrigger(Boolean commitTrigger) { - this.executionSpec = - ExecutionPeersDTO.normalize(executionSpec().withCommitTrigger(commitTrigger)); + @Override + @Nullable + public Long minIntervalMs() { + return minIntervalMs; } - @JsonProperty("minIntervalMs") - private void setMinIntervalMs(Long minIntervalMs) { - this.executionSpec = - ExecutionPeersDTO.normalize(executionSpec().withMinIntervalMs(minIntervalMs)); + @Override + public Map<String, String> jobOptions() { + return ExecutionPeersDTO.unmodifiableOptions(jobOptions); } - @JsonProperty("jobOptions") - private void setJobOptions(Map<String, String> jobOptions) { - this.executionSpec = ExecutionPeersDTO.normalize(executionSpec().withJobOptions(jobOptions)); + @Override + public Map<String, String> updaterOptions() { + return ExecutionPeersDTO.unmodifiableOptions(updaterOptions); } - @JsonProperty("updaterOptions") - private void setUpdaterOptions(Map<String, String> updaterOptions) { - this.executionSpec = - ExecutionPeersDTO.normalize(executionSpec().withUpdaterOptions(updaterOptions)); + @EqualsAndHashCode.Include + private PolicyExecutionSpec executionSpecForEquals() { + return PolicyExecutionSpec.of( + timeTrigger, commitTrigger, minIntervalMs, jobOptions, updaterOptions); } @Override @@ -860,6 +972,32 @@ public interface PolicyContentDTO extends PolicyContent { return PolicyContents.icebergSnapshotExpiration( olderThanDays(), retainLast(), streamResults()); } + + /** Lombok builder customization: unpack PolicyExecutionSpec into flat peer fields. */ + public static class IcebergSnapshotExpirationContentDTOBuilder { + /** + * @param executionSpec peers to copy; {@code null} or empty leaves peer fields unset + * @return this builder + */ + public IcebergSnapshotExpirationContentDTOBuilder withExecutionSpec( + @Nullable PolicyExecutionSpec executionSpec) { + if (executionSpec == null || !executionSpec.hasAny()) { + return this; + } + this.timeTrigger = executionSpec.timeTrigger(); + this.commitTrigger = executionSpec.commitTrigger(); + this.minIntervalMs = executionSpec.minIntervalMs(); + this.jobOptions = + executionSpec.jobOptions().isEmpty() + ? null + : new LinkedHashMap<>(executionSpec.jobOptions()); + this.updaterOptions = + executionSpec.updaterOptions().isEmpty() + ? null + : new LinkedHashMap<>(executionSpec.updaterOptions()); + return this; + } + } } /** Represents typed row-filter policy content. */
