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 708557ed07 [#14227] refactor(policy): Align PolicyExecutionSpec with
module value-object conventions (#14228)
708557ed07 is described below
commit 708557ed07dcef6afb82664f82526c01b2b66352
Author: MaSai <[email protected]>
AuthorDate: Mon Oct 26 16:50:00 2026 +0800
[#14227] refactor(policy): Align PolicyExecutionSpec with module
value-object conventions (#14228)
### What changes were proposed in this pull request?
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
### Why are the changes needed?
Keep the TMS execution-peer value type consistent with the rest of the
`api` policy package.
Fix: #14227
### Does this PR introduce _any_ user-facing change?
Yes — breaking for callers of removed `PolicyExecutionSpec.with*` /
`validate()`. Wire JSON peers unchanged.
### How was this patch tested?
```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 | 36 +-
.../gravitino/policy/TestPolicyExecutionSpec.java | 53 ++-
.../gravitino/dto/policy/PolicyContentDTO.java | 495 ++++++++++++++-------
5 files changed, 423 insertions(+), 286 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 5d12087b5d..62d6d57536 100644
--- a/api/src/test/java/org/apache/gravitino/policy/TestPolicyContents.java
+++ b/api/src/test/java/org/apache/gravitino/policy/TestPolicyContents.java
@@ -174,25 +174,25 @@ public class TestPolicyContents {
}
@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);
-
+ 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"));
}
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 7fbd4bc365..f8c7bb9931 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
@@ -46,88 +46,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));
}
}
@@ -177,7 +152,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 {
@@ -206,12 +181,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() {}
@@ -303,38 +299,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
@@ -378,11 +373,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 {
@@ -395,12 +416,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() {}
@@ -429,38 +471,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
@@ -502,20 +543,68 @@ 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;
+ }
+ }
}
/** Placeholder DTO for Iceberg rewrite-manifests policy content. */
@EqualsAndHashCode
- @ToString(exclude = {"executionSpec"})
+ @ToString(exclude = {"jobOptions", "updaterOptions"})
@Builder(setterPrefix = "with")
@AllArgsConstructor(access = lombok.AccessLevel.PRIVATE)
class IcebergRewriteManifestsContentDTO implements PolicyContentDTO,
ExecutionPeersDTO {
- @EqualsAndHashCode.Exclude @JsonIgnore @Nullable private
PolicyExecutionSpec executionSpec;
- @EqualsAndHashCode.Include
- private PolicyExecutionSpec executionSpecForEquals() {
- return ExecutionPeersDTO.normalize(executionSpec);
- }
+ @EqualsAndHashCode.Exclude
+ @Nullable
+ @JsonProperty("timeTrigger")
+ @JsonInclude(JsonInclude.Include.NON_NULL)
+ private String timeTrigger;
+
+ @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() {}
@@ -527,38 +616,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
@@ -594,20 +682,68 @@ 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;
+ }
+ }
}
/** Placeholder DTO for Iceberg snapshot-expiration policy content. */
@EqualsAndHashCode
- @ToString(exclude = {"executionSpec"})
+ @ToString(exclude = {"jobOptions", "updaterOptions"})
@Builder(setterPrefix = "with")
@AllArgsConstructor(access = lombok.AccessLevel.PRIVATE)
class IcebergSnapshotExpirationContentDTO implements PolicyContentDTO,
ExecutionPeersDTO {
- @EqualsAndHashCode.Exclude @JsonIgnore @Nullable private
PolicyExecutionSpec executionSpec;
- @EqualsAndHashCode.Include
- private PolicyExecutionSpec executionSpecForEquals() {
- return ExecutionPeersDTO.normalize(executionSpec);
- }
+ @EqualsAndHashCode.Exclude
+ @Nullable
+ @JsonProperty("timeTrigger")
+ @JsonInclude(JsonInclude.Include.NON_NULL)
+ private String timeTrigger;
+
+ @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 IcebergSnapshotExpirationContentDTO() {}
@@ -619,38 +755,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
@@ -686,6 +821,32 @@ public interface PolicyContentDTO extends PolicyContent {
spec.jobOptions(),
spec.updaterOptions());
}
+
+ /** 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. */
