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
The following commit(s) were added to refs/heads/main by this push:
new 00eafa5ed9 [MINOR] feat(policy): Persist snapshot-expiration fields
and validate Iceberg policy content (#14229)
00eafa5ed9 is described below
commit 00eafa5ed9a4dd9ef1e6843d14f96ea2a653b47c
Author: MaSai <[email protected]>
AuthorDate: Mon Oct 26 19:38:23 2026 +0800
[MINOR] feat(policy): Persist snapshot-expiration fields and validate
Iceberg policy content (#14229)
### What changes were proposed in this pull request?
1. **Snapshot expiration fields** — add `olderThanDays` (default **5**),
`retainLast` (1), `streamResults` (false) so empty placeholder content
can be persisted without changing `JsonUtils`.
2. **Optimizer wiring (orphan-style)** —
`SnapshotExpirationStrategyHandler` (`trigger-expr: true`),
`SnapshotExpirationJobContext`, `GravitinoSnapshotExpirationJobAdapter`,
registered in `GravitinoJobSubmitter`.
3. **Save-time validation for Iceberg maintenance policies:**
- **compaction** — allowlist `rewriteOptions` names/types/ranges
- **orphan** — reject unsafe `location` syntax (`..`, `\`, `%`,
query/fragment)
- **snapshot expiration** — range-check `olderThanDays` / `retainLast`
- **rewrite manifests** — existing threshold checks (unchanged)
### Why are the changes needed?
- Empty snapshot-expiration content fails Jackson `FAIL_ON_EMPTY_BEANS`
on insert (HTTP 500).
- Compaction accepts unknown/invalid `rewriteOptions` on save; jobs fail
later.
- Snapshot expiration had no recommender path; wire it like orphan
(schedule-driven, always-eligible trigger) rather than a snapshot-count
gate.
### Does this PR introduce _any_ user-facing change?
Yes. Snapshot-expiration content may include the three optional fields;
invalid compaction `rewriteOptions` and unsafe orphan `location` syntax
return **400** on create/update. Operators can register
`gravitino.optimizer.strategyHandler.iceberg-snapshot-expiration.className`
to enable policy-driven expire.
### How was this patch tested?
`./gradlew :api:test --tests
org.apache.gravitino.policy.TestPolicyContents --tests
org.apache.gravitino.policy.TestIcebergOrphanFileRemovalContent --tests
org.apache.gravitino.policy.TestIcebergRewriteManifestsContent --tests
org.apache.gravitino.policy.TestIcebergSnapshotExpirationContent
:maintenance:optimizer:test --tests
org.apache.gravitino.maintenance.optimizer.recommender.job.TestGravitinoSnapshotExpirationJobAdapter
-PskipITs`
---------
Co-authored-by: Cursor <[email protected]>
---
.../policy/IcebergDataCompactionContent.java | 138 ++++-
.../policy/IcebergOrphanFileRemovalContent.java | 13 +
.../policy/IcebergSnapshotExpirationContent.java | 125 ++++-
.../apache/gravitino/policy/PolicyContents.java | 17 +-
.../TestIcebergOrphanFileRemovalContent.java | 18 +
.../TestIcebergSnapshotExpirationContent.java | 70 +++
.../gravitino/policy/TestPolicyContents.java | 28 +
.../gravitino/dto/policy/PolicyContentDTO.java | 74 ++-
.../dto/requests/PolicyUpdateRequest.java | 7 +-
.../apache/gravitino/dto/util/DTOConverters.java | 12 +-
.../policy/TestSnapshotExpirationPolicyDTO.java | 196 +++++++
conf/gravitino-optimizer.conf.template | 6 +
.../org/apache/gravitino/meta/PolicyEntity.java | 38 +-
.../org/apache/gravitino/policy/PolicyManager.java | 12 +-
.../storage/relational/utils/POConverters.java | 4 +-
.../apache/gravitino/policy/TestPolicyManager.java | 38 ++
.../storage/relational/utils/TestPOConverters.java | 41 ++
.../iceberg-expire-snapshots-maintenance-job.md | 578 +++++++--------------
docs/open-api/policies.yaml | 24 +-
.../optimizer-configuration.md | 47 +-
.../optimizer/recommender/Recommender.java | 22 +-
.../snapshot/SnapshotExpirationJobContext.java | 59 +++
.../SnapshotExpirationStrategyHandler.java | 63 +++
.../recommender/job/GravitinoJobSubmitter.java | 5 +-
.../job/GravitinoSnapshotExpirationJobAdapter.java | 120 +++++
.../TestGravitinoSnapshotExpirationJobAdapter.java | 149 ++++++
26 files changed, 1450 insertions(+), 454 deletions(-)
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..011aec1ba1 100644
---
a/api/src/main/java/org/apache/gravitino/policy/IcebergDataCompactionContent.java
+++
b/api/src/main/java/org/apache/gravitino/policy/IcebergDataCompactionContent.java
@@ -27,6 +27,7 @@ import java.util.Locale;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
+import java.util.function.Consumer;
import java.util.regex.Pattern;
import org.apache.commons.lang3.StringUtils;
import org.apache.gravitino.MetadataObject;
@@ -96,9 +97,40 @@ public class IcebergDataCompactionContent implements
PolicyContent {
/** Default Iceberg sort order (empty means unset). */
public static final String DEFAULT_SORT_ORDER = "";
+ private enum RewriteOptionValueKind {
+ POSITIVE_LONG,
+ POSITIVE_INT,
+ NON_NEGATIVE_INT,
+ BOOLEAN,
+ REWRITE_JOB_ORDER,
+ RATIO
+ }
+
private static final Set<String> SUPPORTED_REWRITE_STRATEGIES =
ImmutableSet.of(REWRITE_STRATEGY_BINPACK, REWRITE_STRATEGY_SORT);
private static final Pattern OPTION_KEY_PATTERN =
Pattern.compile("[A-Za-z0-9._-]+");
+ private static final Set<String> REWRITE_JOB_ORDERS =
+ ImmutableSet.of("bytes-asc", "bytes-desc", "files-asc", "files-desc",
"none");
+ // Names mirror Iceberg RewriteDataFiles / SizeBasedFileRewriter /
SizeBasedDataRewriter.
+ private static final Map<String, RewriteOptionValueKind>
SUPPORTED_REWRITE_OPTIONS =
+ ImmutableMap.<String, RewriteOptionValueKind>builder()
+ .put("target-file-size-bytes", RewriteOptionValueKind.POSITIVE_LONG)
+ .put("min-file-size-bytes", RewriteOptionValueKind.POSITIVE_LONG)
+ .put("max-file-size-bytes", RewriteOptionValueKind.POSITIVE_LONG)
+ .put("max-file-group-size-bytes",
RewriteOptionValueKind.POSITIVE_LONG)
+ .put("min-input-files", RewriteOptionValueKind.POSITIVE_INT)
+ .put("max-concurrent-file-group-rewrites",
RewriteOptionValueKind.POSITIVE_INT)
+ .put("partial-progress.max-commits",
RewriteOptionValueKind.POSITIVE_INT)
+ .put("partial-progress.max-failed-commits",
RewriteOptionValueKind.NON_NEGATIVE_INT)
+ .put("output-spec-id", RewriteOptionValueKind.NON_NEGATIVE_INT)
+ .put("delete-file-threshold",
RewriteOptionValueKind.NON_NEGATIVE_INT)
+ .put("rewrite-all", RewriteOptionValueKind.BOOLEAN)
+ .put("partial-progress.enabled", RewriteOptionValueKind.BOOLEAN)
+ .put("use-starting-sequence-number", RewriteOptionValueKind.BOOLEAN)
+ .put("remove-dangling-deletes", RewriteOptionValueKind.BOOLEAN)
+ .put("rewrite-job-order", RewriteOptionValueKind.REWRITE_JOB_ORDER)
+ .put("delete-ratio-threshold", RewriteOptionValueKind.RATIO)
+ .build();
private static final Set<MetadataObject.Type> SUPPORTED_OBJECT_TYPES =
ImmutableSet.of(
MetadataObject.Type.CATALOG, MetadataObject.Type.SCHEMA,
MetadataObject.Type.TABLE);
@@ -284,6 +316,7 @@ public class IcebergDataCompactionContent implements
PolicyContent {
StringUtils.isNotBlank(sortOrder),
"sortOrder must be set when rewriteStrategy is '%s'",
REWRITE_STRATEGY_SORT);
+ // Full Iceberg parse needs Spark ExtendedParser + table Schema; that
runs in the job.
} else {
Preconditions.checkArgument(
StringUtils.isBlank(sortOrder),
@@ -291,21 +324,7 @@ public class IcebergDataCompactionContent implements
PolicyContent {
rewriteStrategy);
}
- rewriteOptions.forEach(
- (key, value) -> {
- Preconditions.checkArgument(StringUtils.isNotBlank(key), "rewrite
option key is blank");
- Preconditions.checkArgument(
- OPTION_KEY_PATTERN.matcher(key).matches(),
- "rewrite option key '%s' contains illegal characters",
- key);
- Preconditions.checkArgument(
- !key.startsWith(JOB_OPTIONS_PREFIX),
- "rewrite option key '%s' must not start with '%s'",
- key,
- JOB_OPTIONS_PREFIX);
- Preconditions.checkArgument(
- StringUtils.isNotBlank(value), "rewrite option '%s' must have
non-empty value", key);
- });
+
rewriteOptions.forEach(IcebergDataCompactionContent::validateRewriteOption);
}
@Override
@@ -360,4 +379,93 @@ public class IcebergDataCompactionContent implements
PolicyContent {
+ rewriteOptions
+ '}';
}
+
+ private static void validateRewriteOption(String key, String value) {
+ Preconditions.checkArgument(StringUtils.isNotBlank(key), "rewrite option
key is blank");
+ Preconditions.checkArgument(
+ OPTION_KEY_PATTERN.matcher(key).matches(),
+ "rewrite option key '%s' contains illegal characters",
+ key);
+ Preconditions.checkArgument(
+ !key.startsWith(JOB_OPTIONS_PREFIX),
+ "rewrite option key '%s' must not start with '%s'",
+ key,
+ JOB_OPTIONS_PREFIX);
+ Preconditions.checkArgument(
+ StringUtils.isNotBlank(value), "rewrite option '%s' must have
non-empty value", key);
+ RewriteOptionValueKind kind = SUPPORTED_REWRITE_OPTIONS.get(key);
+ Preconditions.checkArgument(
+ kind != null,
+ "rewrite option '%s' is not supported by rewrite_data_files; supported
options are %s",
+ key,
+ SUPPORTED_REWRITE_OPTIONS.keySet());
+ switch (kind) {
+ case POSITIVE_LONG:
+ parseRewriteOptionLong(
+ key,
+ value,
+ v -> Preconditions.checkArgument(v > 0, rewriteOptionMessage(key,
"a positive long")));
+ break;
+ case POSITIVE_INT:
+ parseRewriteOptionInt(
+ key,
+ value,
+ v -> Preconditions.checkArgument(v > 0, rewriteOptionMessage(key,
"a positive int")));
+ break;
+ case NON_NEGATIVE_INT:
+ parseRewriteOptionInt(
+ key,
+ value,
+ v ->
+ Preconditions.checkArgument(
+ v >= 0, rewriteOptionMessage(key, "a non-negative int")));
+ break;
+ case BOOLEAN:
+ Preconditions.checkArgument(
+ "true".equalsIgnoreCase(value) || "false".equalsIgnoreCase(value),
+ rewriteOptionMessage(key, "true or false"));
+ break;
+ case REWRITE_JOB_ORDER:
+ Preconditions.checkArgument(
+ REWRITE_JOB_ORDERS.contains(value.toLowerCase(Locale.ROOT)),
+ "rewrite option '%s' must be one of %s, but got '%s'",
+ key,
+ REWRITE_JOB_ORDERS,
+ value);
+ break;
+ case RATIO:
+ try {
+ double ratio = Double.parseDouble(value.trim());
+ Preconditions.checkArgument(
+ ratio >= 0.0D && ratio <= 1.0D,
+ rewriteOptionMessage(key, "a number between 0 and 1"));
+ } catch (NumberFormatException e) {
+ throw new IllegalArgumentException(
+ rewriteOptionMessage(key, "a number between 0 and 1"), e);
+ }
+ break;
+ default:
+ throw new IllegalStateException("Unhandled rewrite option kind: " +
kind);
+ }
+ }
+
+ private static void parseRewriteOptionLong(String key, String value,
Consumer<Long> check) {
+ try {
+ check.accept(Long.parseLong(value.trim()));
+ } catch (NumberFormatException e) {
+ throw new IllegalArgumentException(rewriteOptionMessage(key, "a long"),
e);
+ }
+ }
+
+ private static void parseRewriteOptionInt(String key, String value,
Consumer<Integer> check) {
+ try {
+ check.accept(Integer.parseInt(value.trim()));
+ } catch (NumberFormatException e) {
+ throw new IllegalArgumentException(rewriteOptionMessage(key, "an int"),
e);
+ }
+ }
+
+ private static String rewriteOptionMessage(String key, String expected) {
+ return String.format("rewrite option '%s' must be %s", key, expected);
+ }
}
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..d10f6e2071 100644
---
a/api/src/main/java/org/apache/gravitino/policy/IcebergOrphanFileRemovalContent.java
+++
b/api/src/main/java/org/apache/gravitino/policy/IcebergOrphanFileRemovalContent.java
@@ -143,6 +143,19 @@ public class IcebergOrphanFileRemovalContent implements
PolicyContent {
Preconditions.checkArgument(
location == null ||
location.equals(StringUtils.strip(location.trim())),
"location must not have leading or trailing whitespace");
+ if (location != null) {
+ // Mirror OrphanFileLocationUtils.normalizeLocation safety without
needing a table root.
+ Preconditions.checkArgument(
+ !location.contains(".."), "location must not contain '..' path
segments: %s", location);
+ Preconditions.checkArgument(
+ !location.contains("\\"), "location must not contain '\\': %s",
location);
+ Preconditions.checkArgument(
+ !location.contains("%"), "location must not contain
percent-encoding: %s", location);
+ Preconditions.checkArgument(
+ !location.contains("?") && !location.contains("#"),
+ "location must not contain query or fragment: %s",
+ location);
+ }
}
@Override
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..6c6a62f2df 100644
---
a/api/src/main/java/org/apache/gravitino/policy/IcebergSnapshotExpirationContent.java
+++
b/api/src/main/java/org/apache/gravitino/policy/IcebergSnapshotExpirationContent.java
@@ -18,19 +18,23 @@
*/
package org.apache.gravitino.policy;
+import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.Map;
+import java.util.Objects;
import java.util.Set;
+import javax.annotation.Nullable;
import org.apache.gravitino.MetadataObject;
/**
- * Placeholder built-in policy content for Iceberg snapshot expiration.
+ * Configuration for Iceberg snapshot expiration on an optimizer invocation.
*
- * <p>Configurable thresholds will be added when the expire-snapshots policy
lands; this class
- * registers the built-in type for TMS task mapping.
+ * <p>Maps to {@code builtin-iceberg-expire-snapshots} job options ({@code
older_than} is derived
+ * from {@link #olderThanDays()} by the job adapter; {@code retain_last} and
{@code stream_results}
+ * are forwarded).
*/
public class IcebergSnapshotExpirationContent implements PolicyContent {
/** Property key for strategy type. */
@@ -41,21 +45,68 @@ public class IcebergSnapshotExpirationContent implements
PolicyContent {
public static final String TRIGGER_EXPR_KEY = "trigger-expr";
/** Rule key for the score expression. */
public static final String SCORE_EXPR_KEY = "score-expr";
+ /** Prefix for policy rules forwarded as job options. */
+ public static final String JOB_OPTIONS_PREFIX = "job.options.";
+ /** Retention age option key shared with the job adapter. */
+ public static final String OLDER_THAN_DAYS_KEY = "olderThanDays";
+ /** Retain-last option key shared with the job adapter. */
+ public static final String RETAIN_LAST_KEY = "retainLast";
+ /** Stream-results option key shared with the job adapter. */
+ public static final String STREAM_RESULTS_KEY = "streamResults";
+ /** Maximum retention, approximately 100 years. */
+ public static final long MAX_OLDER_THAN_DAYS = 36500;
/** Strategy type for snapshot expiration. */
public static final String STRATEGY_TYPE_VALUE =
"iceberg-snapshot-expiration";
/** Built-in job template name for expire snapshots. */
public static final String JOB_TEMPLATE_NAME_VALUE =
"builtin-iceberg-expire-snapshots";
+ /** Default retention in days (Iceberg expire_snapshots default: five days
ago). */
+ public static final long DEFAULT_OLDER_THAN_DAYS = 5;
+ /** Default number of recent snapshots to keep. */
+ public static final long DEFAULT_RETAIN_LAST = 1;
+ /** Default stream-results flag. */
+ public static final boolean DEFAULT_STREAM_RESULTS = false;
- private static final Set<MetadataObject.Type> SUPPORTED_OBJECT_TYPES =
- ImmutableSet.of(
- MetadataObject.Type.CATALOG, MetadataObject.Type.SCHEMA,
MetadataObject.Type.TABLE);
+ private final long olderThanDays;
+ private final long retainLast;
+ private final boolean streamResults;
- /** Default constructor for Jackson deserialization only. */
- public IcebergSnapshotExpirationContent() {}
+ /** Creates content with built-in defaults. */
+ public IcebergSnapshotExpirationContent() {
+ this(DEFAULT_OLDER_THAN_DAYS, DEFAULT_RETAIN_LAST, DEFAULT_STREAM_RESULTS);
+ }
+
+ IcebergSnapshotExpirationContent(
+ @Nullable Long olderThanDays, @Nullable Long retainLast, @Nullable
Boolean streamResults) {
+ this.olderThanDays = olderThanDays == null ? DEFAULT_OLDER_THAN_DAYS :
olderThanDays;
+ this.retainLast = retainLast == null ? DEFAULT_RETAIN_LAST : retainLast;
+ this.streamResults = streamResults == null ? DEFAULT_STREAM_RESULTS :
streamResults;
+ }
+
+ /**
+ * @return minimum age of snapshots eligible for expiration, in days
+ */
+ public long olderThanDays() {
+ return olderThanDays;
+ }
+
+ /**
+ * @return minimum number of recent snapshots to keep regardless of age
+ */
+ public long retainLast() {
+ return retainLast;
+ }
+
+ /**
+ * @return whether the expire job should stream intermediate delete results
+ */
+ public boolean streamResults() {
+ return streamResults;
+ }
@Override
public Set<MetadataObject.Type> supportedObjectTypes() {
- return SUPPORTED_OBJECT_TYPES;
+ return ImmutableSet.of(
+ MetadataObject.Type.CATALOG, MetadataObject.Type.SCHEMA,
MetadataObject.Type.TABLE);
}
@Override
@@ -67,24 +118,70 @@ public class IcebergSnapshotExpirationContent implements
PolicyContent {
@Override
public Map<String, Object> rules() {
Map<String, Object> rules = new LinkedHashMap<>();
- // Placeholders must not auto-trigger until real thresholds land in a
follow-up.
- rules.put(TRIGGER_EXPR_KEY, "false");
+ // Eligibility is evaluated when the optimizer is invoked. Scheduling is
owned by the caller.
+ rules.put(TRIGGER_EXPR_KEY, "true");
rules.put(SCORE_EXPR_KEY, "1");
+ rules.put(JOB_OPTIONS_PREFIX + OLDER_THAN_DAYS_KEY, olderThanDays);
+ rules.put(JOB_OPTIONS_PREFIX + RETAIN_LAST_KEY, retainLast);
+ rules.put(JOB_OPTIONS_PREFIX + STREAM_RESULTS_KEY, streamResults);
return Collections.unmodifiableMap(rules);
}
+ /**
+ * Validates retention age before storing a policy or submitting a job.
+ *
+ * @param days minimum snapshot age in days
+ * @throws IllegalArgumentException if days is outside the supported range
+ */
+ public static void validateOlderThanDays(long days) {
+ Preconditions.checkArgument(
+ days >= 1 && days <= MAX_OLDER_THAN_DAYS,
+ "olderThanDays must be between 1 and %s",
+ MAX_OLDER_THAN_DAYS);
+ }
+
+ /**
+ * Validates retain-last before storing a policy or submitting a job.
+ *
+ * @param retainLast number of recent snapshots to keep
+ * @throws IllegalArgumentException if retainLast is not a positive long
+ */
+ public static void validateRetainLast(long retainLast) {
+ Preconditions.checkArgument(retainLast >= 1, "retainLast must be a
positive long");
+ }
+
+ @Override
+ public void validate() {
+ PolicyContent.super.validate();
+ validateOlderThanDays(olderThanDays);
+ validateRetainLast(retainLast);
+ }
+
@Override
public boolean equals(Object other) {
- return other instanceof IcebergSnapshotExpirationContent;
+ if (!(other instanceof IcebergSnapshotExpirationContent)) {
+ return false;
+ }
+ IcebergSnapshotExpirationContent that = (IcebergSnapshotExpirationContent)
other;
+ return olderThanDays == that.olderThanDays
+ && retainLast == that.retainLast
+ && streamResults == that.streamResults;
}
@Override
public int hashCode() {
- return IcebergSnapshotExpirationContent.class.hashCode();
+ return Objects.hash(olderThanDays, retainLast, streamResults);
}
@Override
public String toString() {
- return "IcebergSnapshotExpirationContent{}";
+ return "IcebergSnapshotExpirationContent{"
+ + "olderThanDays="
+ + olderThanDays
+ + ", retainLast="
+ + retainLast
+ + ", streamResults="
+ + streamResults
+ + '}';
}
}
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 d1028a29fb..640424a64b 100644
--- a/api/src/main/java/org/apache/gravitino/policy/PolicyContents.java
+++ b/api/src/main/java/org/apache/gravitino/policy/PolicyContents.java
@@ -206,12 +206,25 @@ public class PolicyContents {
}
/**
- * Creates placeholder Iceberg snapshot-expiration policy content.
+ * Creates Iceberg snapshot-expiration policy content with built-in defaults.
*
* @return snapshot-expiration policy content
*/
public static PolicyContent icebergSnapshotExpiration() {
- return new IcebergSnapshotExpirationContent();
+ return icebergSnapshotExpiration(null, null, null);
+ }
+
+ /**
+ * Creates Iceberg snapshot-expiration policy content. Null arguments use
built-in defaults.
+ *
+ * @param olderThanDays minimum snapshot age in days, or null for the default
+ * @param retainLast minimum recent snapshots to keep, or null for the
default
+ * @param streamResults whether to stream intermediate delete results, or
null for the default
+ * @return snapshot-expiration policy content
+ */
+ public static IcebergSnapshotExpirationContent icebergSnapshotExpiration(
+ @Nullable Long olderThanDays, @Nullable Long retainLast, @Nullable
Boolean streamResults) {
+ return new IcebergSnapshotExpirationContent(olderThanDays, retainLast,
streamResults);
}
/**
diff --git
a/api/src/test/java/org/apache/gravitino/policy/TestIcebergOrphanFileRemovalContent.java
b/api/src/test/java/org/apache/gravitino/policy/TestIcebergOrphanFileRemovalContent.java
index f082ecaad9..87cea9eeda 100644
---
a/api/src/test/java/org/apache/gravitino/policy/TestIcebergOrphanFileRemovalContent.java
+++
b/api/src/test/java/org/apache/gravitino/policy/TestIcebergOrphanFileRemovalContent.java
@@ -94,4 +94,22 @@ class TestIcebergOrphanFileRemovalContent {
Assertions.assertTrue(error.getMessage().contains("olderThanDays"));
}
}
+
+ @Test
+ void rejectsUnsafeLocationSyntax() {
+ for (String location :
+ new String[] {
+ "s3://bucket/table/../other",
+ "s3://bucket/table/%2e%2e/other",
+ "s3://bucket/table\\data",
+ "s3://bucket/table/data?query",
+ "s3://bucket/table/data#frag"
+ }) {
+ IllegalArgumentException error =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> PolicyContents.icebergOrphanFileRemoval(3, location,
true).validate());
+ Assertions.assertTrue(error.getMessage().contains("location"));
+ }
+ }
}
diff --git
a/api/src/test/java/org/apache/gravitino/policy/TestIcebergSnapshotExpirationContent.java
b/api/src/test/java/org/apache/gravitino/policy/TestIcebergSnapshotExpirationContent.java
new file mode 100644
index 0000000000..96ce1c6b4c
--- /dev/null
+++
b/api/src/test/java/org/apache/gravitino/policy/TestIcebergSnapshotExpirationContent.java
@@ -0,0 +1,70 @@
+/*
+ * 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 static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import org.junit.jupiter.api.Test;
+
+/** Unit tests for {@link IcebergSnapshotExpirationContent}. */
+public class TestIcebergSnapshotExpirationContent {
+
+ @Test
+ public void testDefaultsAndRules() {
+ IcebergSnapshotExpirationContent content =
+ (IcebergSnapshotExpirationContent)
PolicyContents.icebergSnapshotExpiration();
+ assertEquals(IcebergSnapshotExpirationContent.DEFAULT_OLDER_THAN_DAYS,
content.olderThanDays());
+ assertEquals(IcebergSnapshotExpirationContent.DEFAULT_RETAIN_LAST,
content.retainLast());
+ assertEquals(IcebergSnapshotExpirationContent.DEFAULT_STREAM_RESULTS,
content.streamResults());
+ assertEquals("true",
content.rules().get(IcebergSnapshotExpirationContent.TRIGGER_EXPR_KEY));
+ assertEquals(
+ content.olderThanDays(),
+ content
+ .rules()
+ .get(
+ IcebergSnapshotExpirationContent.JOB_OPTIONS_PREFIX
+ + IcebergSnapshotExpirationContent.OLDER_THAN_DAYS_KEY));
+ }
+
+ @Test
+ public void testValidateRejectsInvalidBounds() {
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> PolicyContents.icebergSnapshotExpiration(0L, 1L,
false).validate());
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> PolicyContents.icebergSnapshotExpiration(1L, 0L,
false).validate());
+ IllegalArgumentException tooOld =
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ PolicyContents.icebergSnapshotExpiration(
+ IcebergSnapshotExpirationContent.MAX_OLDER_THAN_DAYS +
1, 1L, false)
+ .validate());
+ assertTrue(tooOld.getMessage().contains("olderThanDays"));
+ }
+
+ @Test
+ public void testValidateAcceptsDefaults() {
+ PolicyContents.icebergSnapshotExpiration().validate();
+ PolicyContents.icebergSnapshotExpiration(5L, 1L, true).validate();
+ }
+}
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 ca456c2c05..e378f4c7fd 100644
--- a/api/src/test/java/org/apache/gravitino/policy/TestPolicyContents.java
+++ b/api/src/test/java/org/apache/gravitino/policy/TestPolicyContents.java
@@ -179,6 +179,34 @@ public class TestPolicyContents {
Assertions.assertTrue(exception.getMessage().contains("maxPartitionNum"));
}
+ @Test
+ void testIcebergCompactionContentRejectsUnknownOrInvalidRewriteOptions() {
+ IllegalArgumentException unknown =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ PolicyContents.icebergDataCompaction(1000L, 1L,
mapOf("no-such-option", "x"))
+ .validate());
+ Assertions.assertTrue(unknown.getMessage().contains("no-such-option"));
+
+ IllegalArgumentException negative =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ PolicyContents.icebergDataCompaction(
+ 1000L, 1L, mapOf("target-file-size-bytes", "-1"))
+ .validate());
+
Assertions.assertTrue(negative.getMessage().contains("target-file-size-bytes"));
+
+ IllegalArgumentException notBool =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ PolicyContents.icebergDataCompaction(1000L, 1L,
mapOf("rewrite-all", "yes"))
+ .validate());
+ Assertions.assertTrue(notBool.getMessage().contains("rewrite-all"));
+ }
+
@Test
void testAccessControlContent() {
AccessControlContent content =
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 0692a92a24..323d1d3efa 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
@@ -43,6 +43,7 @@ import org.apache.gravitino.policy.ColumnMaskContent;
import org.apache.gravitino.policy.IcebergDataCompactionContent;
import org.apache.gravitino.policy.IcebergOrphanFileRemovalContent;
import org.apache.gravitino.policy.IcebergRewriteManifestsContent;
+import org.apache.gravitino.policy.IcebergSnapshotExpirationContent;
import org.apache.gravitino.policy.PolicyContent;
import org.apache.gravitino.policy.PolicyContents;
import org.apache.gravitino.policy.RowFilterContent;
@@ -473,17 +474,54 @@ public interface PolicyContentDTO extends PolicyContent {
}
}
- /** Placeholder DTO for Iceberg snapshot-expiration policy content. */
+ /**
+ * Typed snapshot-expiration policy content for REST requests and responses.
+ *
+ * <p>{@code updateContent} replaces the whole content object: omitted
fields fall back to
+ * defaults rather than merging with the previous content.
+ */
@EqualsAndHashCode
@ToString
+ @Builder(setterPrefix = "with")
+ @AllArgsConstructor(access = AccessLevel.PRIVATE)
class IcebergSnapshotExpirationContentDTO implements PolicyContentDTO {
+ @JsonProperty("olderThanDays")
+ @JsonDeserialize(using = SnapshotLongDeserializer.class)
+ private Long olderThanDays;
+
+ @JsonProperty("retainLast")
+ @JsonDeserialize(using = SnapshotLongDeserializer.class)
+ private Long retainLast;
+
+ @JsonProperty("streamResults")
+ @JsonDeserialize(using = SnapshotBooleanDeserializer.class)
+ private Boolean streamResults;
+
private IcebergSnapshotExpirationContentDTO() {}
/**
- * @return empty snapshot-expiration content DTO
+ * @return minimum snapshot age in days, defaulting to five
*/
- public static IcebergSnapshotExpirationContentDTO instance() {
- return new IcebergSnapshotExpirationContentDTO();
+ public long olderThanDays() {
+ return olderThanDays == null
+ ? IcebergSnapshotExpirationContent.DEFAULT_OLDER_THAN_DAYS
+ : olderThanDays;
+ }
+
+ /**
+ * @return minimum recent snapshots to keep, defaulting to one
+ */
+ public long retainLast() {
+ return retainLast == null ?
IcebergSnapshotExpirationContent.DEFAULT_RETAIN_LAST : retainLast;
+ }
+
+ /**
+ * @return whether to stream intermediate delete results
+ */
+ public boolean streamResults() {
+ return streamResults == null
+ ? IcebergSnapshotExpirationContent.DEFAULT_STREAM_RESULTS
+ : streamResults;
}
@Override
@@ -507,8 +545,34 @@ public interface PolicyContentDTO extends PolicyContent {
toDomainContent().validate();
}
+ /** Reads a long option without silently coercing fractional values or
strings. */
+ public static class SnapshotLongDeserializer extends
JsonDeserializer<Long> {
+ @Override
+ public Long deserialize(JsonParser parser, DeserializationContext
context)
+ throws IOException {
+ if (!parser.hasToken(JsonToken.VALUE_NUMBER_INT)) {
+ return context.reportInputMismatch(
+ Long.class, "olderThanDays and retainLast must be 64-bit
integers");
+ }
+ return parser.getLongValue();
+ }
+ }
+
+ /** Reads streamResults without silently coercing numbers or strings to
booleans. */
+ public static class SnapshotBooleanDeserializer extends
JsonDeserializer<Boolean> {
+ @Override
+ public Boolean deserialize(JsonParser parser, DeserializationContext
context)
+ throws IOException {
+ if (!parser.hasToken(JsonToken.VALUE_TRUE) &&
!parser.hasToken(JsonToken.VALUE_FALSE)) {
+ return context.reportInputMismatch(Boolean.class, "streamResults
must be a boolean");
+ }
+ return parser.getBooleanValue();
+ }
+ }
+
private PolicyContent toDomainContent() {
- return PolicyContents.icebergSnapshotExpiration();
+ return PolicyContents.icebergSnapshotExpiration(
+ olderThanDays(), retainLast(), streamResults());
}
}
diff --git
a/common/src/main/java/org/apache/gravitino/dto/requests/PolicyUpdateRequest.java
b/common/src/main/java/org/apache/gravitino/dto/requests/PolicyUpdateRequest.java
index 49d4a3b547..0473426043 100644
---
a/common/src/main/java/org/apache/gravitino/dto/requests/PolicyUpdateRequest.java
+++
b/common/src/main/java/org/apache/gravitino/dto/requests/PolicyUpdateRequest.java
@@ -121,7 +121,12 @@ public interface PolicyUpdateRequest extends RESTRequest {
public void validate() throws IllegalArgumentException {}
}
- /** The policy update request for updating a policy content. */
+ /**
+ * The policy update request for updating a policy content.
+ *
+ * <p>{@code newContent} fully replaces the previous content. Omitted typed
fields use that
+ * content type's defaults; they are not merged with the previously stored
values.
+ */
@EqualsAndHashCode
@Getter
@ToString
diff --git
a/common/src/main/java/org/apache/gravitino/dto/util/DTOConverters.java
b/common/src/main/java/org/apache/gravitino/dto/util/DTOConverters.java
index 69f7922280..e6996d9f66 100644
--- a/common/src/main/java/org/apache/gravitino/dto/util/DTOConverters.java
+++ b/common/src/main/java/org/apache/gravitino/dto/util/DTOConverters.java
@@ -700,7 +700,12 @@ public class DTOConverters {
.build();
}
if (policyContent instanceof IcebergSnapshotExpirationContent) {
- return PolicyContentDTO.IcebergSnapshotExpirationContentDTO.instance();
+ IcebergSnapshotExpirationContent content =
(IcebergSnapshotExpirationContent) policyContent;
+ return PolicyContentDTO.IcebergSnapshotExpirationContentDTO.builder()
+ .withOlderThanDays(content.olderThanDays())
+ .withRetainLast(content.retainLast())
+ .withStreamResults(content.streamResults())
+ .build();
}
if (policyContent instanceof IcebergDataCompactionContent) {
IcebergDataCompactionContent icebergCompactionContent =
@@ -1600,7 +1605,10 @@ public class DTOConverters {
content.useCaching());
}
if (policyContentDTO instanceof
PolicyContentDTO.IcebergSnapshotExpirationContentDTO) {
- return PolicyContents.icebergSnapshotExpiration();
+ PolicyContentDTO.IcebergSnapshotExpirationContentDTO content =
+ (PolicyContentDTO.IcebergSnapshotExpirationContentDTO)
policyContentDTO;
+ return PolicyContents.icebergSnapshotExpiration(
+ content.olderThanDays(), content.retainLast(),
content.streamResults());
}
if (policyContentDTO instanceof
PolicyContentDTO.IcebergCompactionContentDTO) {
PolicyContentDTO.IcebergCompactionContentDTO icebergCompactionContentDTO
=
diff --git
a/common/src/test/java/org/apache/gravitino/dto/policy/TestSnapshotExpirationPolicyDTO.java
b/common/src/test/java/org/apache/gravitino/dto/policy/TestSnapshotExpirationPolicyDTO.java
new file mode 100644
index 0000000000..7046c74c23
--- /dev/null
+++
b/common/src/test/java/org/apache/gravitino/dto/policy/TestSnapshotExpirationPolicyDTO.java
@@ -0,0 +1,196 @@
+/*
+ * 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.dto.policy;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import java.time.Instant;
+import org.apache.gravitino.dto.AuditDTO;
+import org.apache.gravitino.dto.requests.PolicyCreateRequest;
+import org.apache.gravitino.dto.requests.PolicyUpdateRequest;
+import org.apache.gravitino.dto.util.DTOConverters;
+import org.apache.gravitino.json.JsonUtils;
+import org.apache.gravitino.policy.IcebergSnapshotExpirationContent;
+import org.apache.gravitino.policy.PolicyContent;
+import org.apache.gravitino.policy.PolicyContents;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+class TestSnapshotExpirationPolicyDTO {
+
+ @Test
+ void createRequestDefaultsAndValidation() throws Exception {
+ String json =
+
"{\"name\":\"expire\",\"policyType\":\"system_iceberg_snapshot_expiration\",\"enabled\":true,\"content\":{}}";
+ PolicyCreateRequest request =
+ JsonUtils.objectMapper().readValue(json, PolicyCreateRequest.class);
+ request.validate();
+ Assertions.assertEquals(
+ PolicyContents.icebergSnapshotExpiration(),
+ DTOConverters.fromDTO(request.getPolicyContent()));
+ String invalid = json.replace("\"content\":{}",
"\"content\":{\"olderThanDays\":0}");
+ PolicyCreateRequest bad =
+ JsonUtils.objectMapper().readValue(invalid, PolicyCreateRequest.class);
+ Assertions.assertThrows(IllegalArgumentException.class, bad::validate);
+ }
+
+ @Test
+ void updateContentRoundTrip() throws Exception {
+ String json =
+
"{\"@type\":\"updateContent\",\"policyType\":\"system_iceberg_snapshot_expiration\","
+ +
"\"newContent\":{\"olderThanDays\":14,\"retainLast\":3,\"streamResults\":true}}";
+ PolicyUpdateRequest request =
+ JsonUtils.objectMapper().readValue(json, PolicyUpdateRequest.class);
+ request.validate();
+ PolicyUpdateRequest.UpdatePolicyContentRequest update =
+ (PolicyUpdateRequest.UpdatePolicyContentRequest) request;
+ Assertions.assertEquals(
+ PolicyContents.icebergSnapshotExpiration(14L, 3L, true),
+ DTOConverters.fromDTO(update.getNewContent()));
+ Assertions.assertEquals(
+ request,
+ JsonUtils.objectMapper()
+ .readValue(
+ JsonUtils.objectMapper().writeValueAsString(request),
PolicyUpdateRequest.class));
+ }
+
+ @Test
+ void updateContentOmitsResetToDefaults() throws Exception {
+ // updateContent fully replaces content; omitted fields use defaults, not
prior values.
+ String json =
+
"{\"@type\":\"updateContent\",\"policyType\":\"system_iceberg_snapshot_expiration\","
+ + "\"newContent\":{\"olderThanDays\":30}}";
+ PolicyUpdateRequest.UpdatePolicyContentRequest update =
+ JsonUtils.objectMapper()
+ .readValue(json,
PolicyUpdateRequest.UpdatePolicyContentRequest.class);
+ update.validate();
+ Assertions.assertEquals(
+ PolicyContents.icebergSnapshotExpiration(
+ 30L,
+ IcebergSnapshotExpirationContent.DEFAULT_RETAIN_LAST,
+ IcebergSnapshotExpirationContent.DEFAULT_STREAM_RESULTS),
+ DTOConverters.fromDTO(update.getNewContent()));
+ }
+
+ @Test
+ void responseRoundTripRetainsTypedContent() throws Exception {
+ PolicyContent domain = PolicyContents.icebergSnapshotExpiration(14L, 5L,
true);
+ PolicyDTO dto =
+ PolicyDTO.builder()
+ .withName("expire")
+ .withPolicyType("system_iceberg_snapshot_expiration")
+ .withEnabled(true)
+ .withContent(DTOConverters.toDTO(domain))
+
.withAudit(AuditDTO.builder().withCreator("test").withCreateTime(Instant.EPOCH).build())
+ .build();
+ PolicyDTO restored =
+ JsonUtils.objectMapper()
+ .readValue(JsonUtils.objectMapper().writeValueAsString(dto),
PolicyDTO.class);
+ Assertions.assertEquals(dto, restored);
+ Assertions.assertEquals(domain, DTOConverters.fromDTO(restored.content()));
+ }
+
+ @ParameterizedTest
+ @CsvSource({
+ "olderThanDays,14.5",
+ "olderThanDays,\"14\"",
+ "olderThanDays,true",
+ "retainLast,1.0",
+ "retainLast,\"1\"",
+ "retainLast,false",
+ "streamResults,0",
+ "streamResults,1",
+ "streamResults,\"true\"",
+ "streamResults,\"false\"",
+ "streamResults,[]",
+ "streamResults,{}"
+ })
+ void rejectInvalidValuesInCreateAndUpdate(String field, String value) {
+ String content = "{\"" + field + "\":" + value + "}";
+ String create =
+
"{\"name\":\"expire\",\"policyType\":\"system_iceberg_snapshot_expiration\","
+ + "\"enabled\":true,\"content\":"
+ + content
+ + "}";
+ String update =
+
"{\"@type\":\"updateContent\",\"policyType\":\"system_iceberg_snapshot_expiration\","
+ + "\"newContent\":"
+ + content
+ + "}";
+ Assertions.assertThrows(
+ JsonProcessingException.class,
+ () -> JsonUtils.objectMapper().readValue(create,
PolicyCreateRequest.class));
+ Assertions.assertThrows(
+ JsonProcessingException.class,
+ () ->
+ JsonUtils.objectMapper()
+ .readValue(update,
PolicyUpdateRequest.UpdatePolicyContentRequest.class));
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void booleanStreamResultsValues(boolean streamResults) throws Exception {
+ PolicyContentDTO.IcebergSnapshotExpirationContentDTO dto =
+ JsonUtils.objectMapper()
+ .readValue(
+ "{\"streamResults\":" + streamResults + "}",
+ PolicyContentDTO.IcebergSnapshotExpirationContentDTO.class);
+ dto.validate();
+ Assertions.assertEquals(
+ PolicyContents.icebergSnapshotExpiration(null, null, streamResults),
+ DTOConverters.fromDTO(dto));
+ }
+
+ @Test
+ void createAndUpdateValidateRetentionUpperBoundary() throws Exception {
+ long maximum = IcebergSnapshotExpirationContent.MAX_OLDER_THAN_DAYS;
+ for (long days : new long[] {maximum, maximum + 1, Long.MAX_VALUE}) {
+ String content = "{\"olderThanDays\":" + days + "}";
+ PolicyCreateRequest create =
+ JsonUtils.objectMapper()
+ .readValue(
+
"{\"name\":\"expire\",\"policyType\":\"system_iceberg_snapshot_expiration\",\"content\":"
+ + content
+ + "}",
+ PolicyCreateRequest.class);
+ PolicyUpdateRequest update =
+ JsonUtils.objectMapper()
+ .readValue(
+
"{\"@type\":\"updateContent\",\"policyType\":\"system_iceberg_snapshot_expiration\",\"newContent\":"
+ + content
+ + "}",
+ PolicyUpdateRequest.class);
+ if (days == maximum) {
+ Assertions.assertDoesNotThrow(create::validate);
+ Assertions.assertDoesNotThrow(update::validate);
+ } else {
+ Assertions.assertTrue(
+ Assertions.assertThrows(IllegalArgumentException.class,
create::validate)
+ .getMessage()
+ .contains("olderThanDays"));
+ Assertions.assertTrue(
+ Assertions.assertThrows(IllegalArgumentException.class,
update::validate)
+ .getMessage()
+ .contains("olderThanDays"));
+ }
+ }
+ }
+}
diff --git a/conf/gravitino-optimizer.conf.template
b/conf/gravitino-optimizer.conf.template
index 6c01b37378..1693c32c52 100644
--- a/conf/gravitino-optimizer.conf.template
+++ b/conf/gravitino-optimizer.conf.template
@@ -33,6 +33,12 @@ gravitino.optimizer.recommender.tableMetaProvider =
gravitino-table-metadata-pro
# The job submitter for the recommender (noop-job-submitter keeps
recommendations local).
gravitino.optimizer.recommender.jobSubmitter = noop-job-submitter
+# Built-in strategy handlers (optional overrides; Recommender has the same
defaults in code).
+# gravitino.optimizer.strategyHandler.iceberg-data-compaction.className =
org.apache.gravitino.maintenance.optimizer.recommender.handler.compaction.CompactionStrategyHandler
+# gravitino.optimizer.strategyHandler.iceberg-rewrite-manifests.className =
org.apache.gravitino.maintenance.optimizer.recommender.handler.ManifestRewriteStrategyHandler
+# gravitino.optimizer.strategyHandler.iceberg-orphan-file-removal.className =
org.apache.gravitino.maintenance.optimizer.recommender.handler.orphan.OrphanFileRemovalStrategyHandler
+# gravitino.optimizer.strategyHandler.iceberg-snapshot-expiration.className =
org.apache.gravitino.maintenance.optimizer.recommender.handler.snapshot.SnapshotExpirationStrategyHandler
+
# Updater plugins (push statistics/metrics back into Gravitino)
# The statistics updater for the recommender/update-statistics tool.
gravitino.optimizer.updater.statisticsUpdater = gravitino-statistics-updater
diff --git a/core/src/main/java/org/apache/gravitino/meta/PolicyEntity.java
b/core/src/main/java/org/apache/gravitino/meta/PolicyEntity.java
index ad9bcf2af3..0fae994c48 100644
--- a/core/src/main/java/org/apache/gravitino/meta/PolicyEntity.java
+++ b/core/src/main/java/org/apache/gravitino/meta/PolicyEntity.java
@@ -160,7 +160,17 @@ public class PolicyEntity implements Entity, Auditable,
HasIdentifier {
@Override
public void validate() throws IllegalArgumentException {
Entity.super.validate();
- validatePolicy();
+ validatePolicy(true /* validateContent */);
+ }
+
+ /**
+ * Validates policy metadata and type/content class matching without calling
{@link
+ * PolicyContent#validate()}. Used when renaming, commenting, or enabling a
policy whose stored
+ * content may predate stricter content checks.
+ */
+ public void validateExceptContent() throws IllegalArgumentException {
+ Entity.super.validate();
+ validatePolicy(false /* validateContent */);
}
@Override
@@ -182,7 +192,7 @@ public class PolicyEntity implements Entity, Auditable,
HasIdentifier {
return Objects.hash(id, name, namespace, policyType, comment, enabled,
content, auditInfo);
}
- private void validatePolicy() {
+ private void validatePolicy(boolean validateContent) {
Preconditions.checkArgument(StringUtils.isNotBlank(name()), "Policy name
cannot be blank");
Preconditions.checkArgument(content() != null, "Policy content cannot be
null");
@@ -198,7 +208,9 @@ public class PolicyEntity implements Entity, Auditable,
HasIdentifier {
policyType.contentClass().getSimpleName(),
policyType.name(),
content().getClass().getName());
- content().validate();
+ if (validateContent) {
+ content().validate();
+ }
}
public static class Builder {
@@ -252,6 +264,24 @@ public class PolicyEntity implements Entity, Auditable,
HasIdentifier {
}
public PolicyEntity build() {
+ PolicyEntity policyEntity = newEntity();
+ policyEntity.validate();
+ return policyEntity;
+ }
+
+ /**
+ * Builds a policy entity without invoking {@link
PolicyContent#validate()}. Prefer {@link
+ * #build()} for create and content-update paths.
+ *
+ * @return policy entity with metadata validated only
+ */
+ public PolicyEntity buildSkippingContentValidation() {
+ PolicyEntity policyEntity = newEntity();
+ policyEntity.validateExceptContent();
+ return policyEntity;
+ }
+
+ private PolicyEntity newEntity() {
PolicyEntity policyEntity = new PolicyEntity();
policyEntity.id = id;
policyEntity.name = name;
@@ -261,8 +291,6 @@ public class PolicyEntity implements Entity, Auditable,
HasIdentifier {
policyEntity.enabled = enabled;
policyEntity.content = content;
policyEntity.auditInfo = auditInfo;
- policyEntity.validate();
-
return policyEntity;
}
}
diff --git a/core/src/main/java/org/apache/gravitino/policy/PolicyManager.java
b/core/src/main/java/org/apache/gravitino/policy/PolicyManager.java
index eb3c8b5848..52d8d91c5c 100644
--- a/core/src/main/java/org/apache/gravitino/policy/PolicyManager.java
+++ b/core/src/main/java/org/apache/gravitino/policy/PolicyManager.java
@@ -263,7 +263,8 @@ public class PolicyManager implements PolicyDispatcher {
policyEntity -> {
PolicyEntity.Builder builder = newPolicyBuilder(policyEntity);
builder.withEnabled(expectedEnabledState);
- return builder.build();
+ // Enable/disable must not re-validate legacy stored content.
+ return builder.buildSkippingContentValidation();
});
} catch (NoSuchEntityException e) {
// The policy was dropped between the state check above and this update.
@@ -314,6 +315,7 @@ public class PolicyManager implements PolicyDispatcher {
String newName = policyEntity.name();
String newComment = policyEntity.comment();
PolicyContent newContent = policyEntity.content();
+ boolean contentUpdated = false;
for (PolicyChange change : changes) {
if (change instanceof PolicyChange.RenamePolicy) {
@@ -343,20 +345,24 @@ public class PolicyManager implements PolicyDispatcher {
}
newContent = updateContent.getContent();
+ contentUpdated = true;
validateReadRestrictionContent(policyType, newContent);
} else {
throw new IllegalArgumentException("Unsupported policy change: " +
change);
}
}
- newContent.validate();
+ // Rename/comment must not re-validate legacy stored content that would
fail new checks.
+ if (contentUpdated) {
+ newContent.validate();
+ }
PolicyEntity.Builder builder = newPolicyBuilder(policyEntity);
builder.withName(newName);
builder.withComment(newComment);
builder.withContent(newContent);
- return builder.build();
+ return contentUpdated ? builder.build() :
builder.buildSkippingContentValidation();
}
private static void validateReadRestrictionContent(
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/utils/POConverters.java
b/core/src/main/java/org/apache/gravitino/storage/relational/utils/POConverters.java
index 9dfe048e2c..b56d38825e 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/utils/POConverters.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/utils/POConverters.java
@@ -1558,6 +1558,8 @@ public class POConverters {
public static PolicyEntity fromPolicyPO(PolicyPO policyPO, Namespace
namespace) {
try {
Policy.BuiltInType policyType =
Policy.BuiltInType.fromPolicyType(policyPO.getPolicyType());
+ // Skip PolicyContent.validate() so stricter checks added after write do
not break load,
+ // rename, comment, or enable of already-stored policies.
Create/UpdateContent still validate.
return PolicyEntity.builder()
.withId(policyPO.getPolicyId())
.withName(policyPO.getPolicyName())
@@ -1570,7 +1572,7 @@ public class POConverters {
.readValue(policyPO.getPolicyVersionPO().getContent(),
policyType.contentClass()))
.withAuditInfo(
JsonUtils.anyFieldMapper().readValue(policyPO.getAuditInfo(),
AuditInfo.class))
- .build();
+ .buildSkippingContentValidation();
} catch (JsonProcessingException e) {
throw new RuntimeException("Failed to deserialize json object:", e);
}
diff --git
a/core/src/test/java/org/apache/gravitino/policy/TestPolicyManager.java
b/core/src/test/java/org/apache/gravitino/policy/TestPolicyManager.java
index 72c9a52d29..51887d8b05 100644
--- a/core/src/test/java/org/apache/gravitino/policy/TestPolicyManager.java
+++ b/core/src/test/java/org/apache/gravitino/policy/TestPolicyManager.java
@@ -588,6 +588,44 @@ public class TestPolicyManager {
Assertions.assertEquals(legacyContent, commented.content());
}
+ @Test
+ public void testRenameDoesNotRevalidateLegacyCompactionRewriteOptions()
throws IOException {
+ String policyName = "legacy_compaction_" +
UUID.randomUUID().toString().replace("-", "");
+ // Constructed without content validate(); would fail rewriteOptions
allowlist on create.
+ PolicyContent legacyContent =
+ PolicyContents.icebergDataCompaction(1000L, 1L,
ImmutableMap.of("no-such-option", "x"));
+ PolicyEntity legacyPolicy =
+ PolicyEntity.builder()
+ .withId(new RandomIdGenerator().nextId())
+ .withName(policyName)
+ .withNamespace(NamespaceUtil.ofPolicy(METALAKE))
+ .withPolicyType(Policy.BuiltInType.ICEBERG_COMPACTION)
+ .withEnabled(true)
+ .withContent(legacyContent)
+ .withAuditInfo(
+
AuditInfo.builder().withCreator("test").withCreateTime(Instant.now()).build())
+ .buildSkippingContentValidation();
+ entityStore.put(legacyPolicy, false /* overwritten */);
+
+ String newName = "renamed_" + policyName;
+ PolicyEntity renamed =
+ Assertions.assertDoesNotThrow(
+ () -> policyManager.alterPolicy(METALAKE, policyName,
PolicyChange.rename(newName)));
+ Assertions.assertEquals(newName, renamed.name());
+ Assertions.assertEquals(legacyContent, renamed.content());
+
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ policyManager.alterPolicy(
+ METALAKE,
+ newName,
+ PolicyChange.updateContent(
+ "system_iceberg_compaction",
+ PolicyContents.icebergDataCompaction(
+ 1000L, 1L, ImmutableMap.of("no-such-option", "y")))));
+ }
+
@Test
public void testCreateUpdateAndGetColumnMaskPolicy() {
String policyName = "column_mask_" +
UUID.randomUUID().toString().replace("-", "");
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..759bbdc059 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;
@@ -1032,6 +1033,46 @@ public class TestPOConverters {
assertEquals(expectedPolicy.content(), convertedPolicy.content());
}
+ @Test
+ public void testFromPolicyPOIcebergSnapshotExpiration() throws
JsonProcessingException {
+ PolicyContent content = PolicyContents.icebergSnapshotExpiration(14L, 5L,
true);
+ String stored = JsonUtils.anyFieldMapper().writeValueAsString(content);
+ assertTrue(stored.contains("\"olderThanDays\":14"));
+ assertTrue(stored.contains("\"retainLast\":5"));
+ assertTrue(stored.contains("\"streamResults\":true"));
+
+ PolicyEntity converted =
+ POConverters.fromPolicyPO(
+ createPolicyPO(
+ 2L,
+ "snapshot-expiration",
+ "system_iceberg_snapshot_expiration",
+ 1L,
+ createPolicyVersionPO(2L, 1L, 1L, "expire", true, content)),
+ NamespaceUtil.ofPolicy("test_metalake"));
+ assertEquals(Policy.BuiltInType.ICEBERG_SNAPSHOT_EXPIRATION,
converted.policyType());
+ assertEquals(content, converted.content());
+ }
+
+ @Test
+ public void testFromPolicyPOIcebergRewriteManifestsPersists() throws
JsonProcessingException {
+ PolicyContent content = PolicyContents.icebergRewriteManifests();
+ String stored = JsonUtils.anyFieldMapper().writeValueAsString(content);
+ assertFalse(stored.isEmpty());
+
+ PolicyEntity converted =
+ POConverters.fromPolicyPO(
+ createPolicyPO(
+ 3L,
+ "rewrite-manifests",
+ "system_iceberg_rewrite_manifests",
+ 1L,
+ createPolicyVersionPO(3L, 1L, 1L, "rewrite", true, content)),
+ NamespaceUtil.ofPolicy("test_metalake"));
+ assertEquals(Policy.BuiltInType.ICEBERG_REWRITE_MANIFESTS,
converted.policyType());
+ assertEquals(content, converted.content());
+ }
+
@Test
public void testFromTagPO() throws JsonProcessingException {
TagPO tagPO = createTagPO(1L, "test", 1L, "this is test");
diff --git a/design-docs/iceberg-expire-snapshots-maintenance-job.md
b/design-docs/iceberg-expire-snapshots-maintenance-job.md
index 483ab3c1c7..b856b4c653 100644
--- a/design-docs/iceberg-expire-snapshots-maintenance-job.md
+++ b/design-docs/iceberg-expire-snapshots-maintenance-job.md
@@ -21,9 +21,10 @@
| Field | Value
|
| ------- |
-------------------------------------------------------------------- |
-| Status | Draft
|
+| Status | Implemented (aligned with orphan cleanup)
|
| Authors | @laserninja
|
| Created | 2026-05-26
|
+| Updated | 2026-10-26
|
| Issue | [#11194](https://github.com/apache/gravitino/issues/11194)
|
| Module | `api`, `maintenance/jobs`, `maintenance/optimizer`
|
@@ -43,24 +44,26 @@ Trino-side delegation for `expire_snapshots` as a
procedure, but there is
no server-side built-in job that can be triggered automatically via the
Table Maintenance Service (Optimizer) policies.
-This design proposes adding full end-to-end support for Iceberg snapshot
-expiration: from policy definition through strategy evaluation to Spark
-job execution.
+This design adds end-to-end support for Iceberg snapshot expiration: from
+policy definition through strategy evaluation to Spark job execution,
+following the same pattern as orphan-file removal.
---
## 2. Goals
-1. Add a new built-in policy type `system_iceberg_snapshot_expiration` for
- declarative snapshot expiration configuration.
-2. Add a strategy handler that evaluates when snapshot expiration should
- run, based on the snapshot count read from Iceberg table metadata.
-3. Expose that snapshot count as a reserved Iceberg table property so the
- handler needs no statistics collection.
-4. Add a job adapter that converts strategy evaluation results into job
- configurations.
-5. Add the Spark job that executes Iceberg's `expire_snapshots` procedure.
-6. Ensure the full flow works end-to-end: policy → strategy → job
+1. Add a built-in policy type `system_iceberg_snapshot_expiration` for
+ declarative retention configuration (`olderThanDays`, `retainLast`,
+ `streamResults`).
+2. Add a strategy handler that follows orphan cleanup: `trigger-expr` is
+ `true`, so each optimizer invocation may submit expire when the policy
+ is attached. Scheduling remains external (optimizer CLI / cron / TMS).
+3. Add a job adapter that maps policy options to
+ `builtin-iceberg-expire-snapshots` jobConf (`older_than`, `retain_last`,
+ `stream_results`).
+4. Reuse the Spark job that executes Iceberg's `expire_snapshots`
+ procedure.
+5. Ensure the full flow works end-to-end: policy → strategy → job
submission → Spark execution.
---
@@ -71,44 +74,47 @@ job execution.
- Automatic policy creation - users must explicitly create and attach
policies.
- Changes to the Optimizer scheduling framework itself.
+- Triggering expire from a `snapshot-count` metadata gate (see
+ [§6](#6-trigger-model)). Industry practice is schedule-driven retention,
+ not count thresholds.
---
## 4. Existing Architecture Overview
The Gravitino maintenance module follows a layered architecture for
-automated table maintenance. The existing Iceberg compaction flow
-establishes the pattern:
+automated table maintenance. Orphan-file removal is the closest precedent
+for whole-table, always-eligible cleanup:
```
Policy Creation (REST API)
↓
GravitinoStrategyProvider (loads policies as strategies)
↓
-CompactionStrategyHandler (evaluates trigger / score expressions)
+OrphanFileRemovalStrategyHandler (BaseExpressionStrategyHandler, trigger-expr
true)
↓
-CompactionJobContext → GravitinoCompactionJobAdapter (converts to job config)
+OrphanFileRemovalJobContext → GravitinoOrphanFileRemovalJobAdapter
↓
GravitinoJobSubmitter (submits job via REST)
↓
-IcebergRewriteDataFilesJob (Spark execution)
+IcebergRemoveOrphanFilesJob (Spark execution)
```
### 4.1 Layer Summary
-| Layer | Compaction Components
| Purpose |
+| Layer | Orphan Components
| Purpose |
| ------------ |
----------------------------------------------------------------------------------
| -------------------------------------------------- |
-| **Policy** | `Policy.BuiltInType.ICEBERG_COMPACTION`,
`IcebergDataCompactionContent` | Define configuration, thresholds,
expressions |
-| **Strategy** | `CompactionStrategyHandler` extends
`BaseExpressionStrategyHandler` | Evaluate trigger conditions,
score partitions |
-| **Adapter** | `GravitinoCompactionJobAdapter`, `CompactionJobContext`
| Convert evaluation result to job configuration |
-| **Job** | `IcebergRewriteDataFilesJob`, registered in
`BuiltInJobTemplateProvider` | Execute Spark procedure
|
+| **Policy** | `ICEBERG_ORPHAN_FILE_REMOVAL`,
`IcebergOrphanFileRemovalContent` | Retention and job options
|
+| **Strategy** | `OrphanFileRemovalStrategyHandler` extends
`BaseExpressionStrategyHandler` | `trigger-expr: true`; table-level
eligibility |
+| **Adapter** | `GravitinoOrphanFileRemovalJobAdapter`,
`OrphanFileRemovalJobContext` | Convert options to jobConf
|
+| **Job** | `IcebergRemoveOrphanFilesJob`
| Execute Spark procedure |
---
## 5. Proposed Design
-We add the same four layers for snapshot expiration, following the
-compaction pattern.
+We add the same four layers for snapshot expiration, mirroring orphan
+cleanup rather than compaction's statistics-driven expressions.
### 5.1 Architecture Diagram
@@ -118,6 +124,7 @@ compaction pattern.
│ type: "system_iceberg_snapshot_expiration" │
│ content: IcebergSnapshotExpirationContent │
│ { olderThanDays, retainLast, streamResults } │
+│ rules: trigger-expr=true, job.options.* │
└──────────────────────────┬───────────────────────────────────┘
↓
┌──────────────────────────────────────────────────────────────┐
@@ -129,17 +136,17 @@ compaction pattern.
↓
┌──────────────────────────────────────────────────────────────┐
│ SnapshotExpirationStrategyHandler │
-│ implements StrategyHandler │
+│ extends BaseExpressionStrategyHandler │
│ dataRequirements: {TABLE_METADATA} │
-│ Reads `snapshot-count` from table metadata properties │
-│ Evaluates: snapshot count ≥ minSnapshotCount │
+│ trigger-expr true → shouldTrigger() when optimizer runs │
│ Returns: StrategyEvaluation with score + context │
└──────────────────────────┬───────────────────────────────────┘
↓
┌──────────────────────────────────────────────────────────────┐
│ SnapshotExpirationJobContext → JobAdapter │
-│ Extracts: older_than, retain_last, stream_results │
-│ Builds: job config map for template substitution │
+│ olderThanDays → older_than (UTC timestamp) │
+│ retainLast → retain_last │
+│ streamResults → stream_results │
└──────────────────────────┬───────────────────────────────────┘
↓
┌──────────────────────────────────────────────────────────────┐
@@ -160,126 +167,57 @@ compaction pattern.
### 5.2 Layer 1 - Policy Definition (`api/`)
-#### 5.2.1 New Policy Type
+#### 5.2.1 Policy Type
-Add `ICEBERG_SNAPSHOT_EXPIRATION` to `Policy.BuiltInType`:
+`ICEBERG_SNAPSHOT_EXPIRATION` / `system_iceberg_snapshot_expiration` maps to
+`IcebergSnapshotExpirationContent`.
-```java
-// api/src/main/java/org/apache/gravitino/policy/Policy.java
-enum BuiltInType {
- ICEBERG_COMPACTION("system_iceberg_compaction",
- IcebergDataCompactionContent.class),
- ICEBERG_SNAPSHOT_EXPIRATION("system_iceberg_snapshot_expiration",
- IcebergSnapshotExpirationContent.class), // NEW
- CUSTOM("custom", CustomContent.class);
-}
-```
-
-#### 5.2.2 New Policy Content Class
-
-Create `IcebergSnapshotExpirationContent` following the
-`IcebergDataCompactionContent` pattern:
+#### 5.2.2 Policy Content
```java
-// NEW: api/src/main/java/org/apache/gravitino/policy/
-// IcebergSnapshotExpirationContent.java
public class IcebergSnapshotExpirationContent implements PolicyContent {
+ public static final String STRATEGY_TYPE_VALUE =
"iceberg-snapshot-expiration";
+ public static final String JOB_TEMPLATE_NAME_VALUE =
"builtin-iceberg-expire-snapshots";
- // Strategy metadata
- public static final String STRATEGY_TYPE_VALUE =
- "iceberg-snapshot-expiration";
- public static final String JOB_TEMPLATE_NAME_VALUE =
- "builtin-iceberg-expire-snapshots";
-
- // Configurable fields - all independently optional
- @Nullable private final Long olderThanDays; // no Gravitino default
- @Nullable private final Long retainLast; // no Gravitino default
- @Nullable private final Boolean streamResults;
- private final long minSnapshotCount; // trigger threshold, default: 10
+ public static final long DEFAULT_OLDER_THAN_DAYS = 5; // Iceberg expire
default
+ public static final long DEFAULT_RETAIN_LAST = 1;
+ public static final boolean DEFAULT_STREAM_RESULTS = false;
- // Rule keys
- public static final String MIN_SNAPSHOT_COUNT_KEY = "minSnapshotCount";
+ private final long olderThanDays;
+ private final long retainLast;
+ private final boolean streamResults;
- // Defaults
- public static final boolean DEFAULT_STREAM_RESULTS = true;
- public static final long DEFAULT_MIN_SNAPSHOT_COUNT = 10;
+ // rules(): trigger-expr=true, score-expr=1,
+ // job.options.olderThanDays / retainLast / streamResults
}
```
-`olderThanDays` and `retainLast` deliberately have **no Gravitino-side
-default**. See [§5.2.3](#523-expiration-parameters-and-defaults).
+Null JSON fields use the defaults above (same pattern as orphan's
+`olderThanDays` / `dryRun`).
#### 5.2.3 Expiration Parameters and Defaults
-Both expiration conditions are **independently optional**. A user may
-specify only `olderThanDays`, only `retainLast`, or both - the job adapter
-omits an unset condition from the procedure call rather than filling it in
-with a Gravitino default. (`streamResults` is different: it is a transport
-knob rather than an expiration condition, so it is always passed - see
-[§5.2.4](#524-why-streamresults-defaults-to-true).)
-
-**Why no Gravitino default for `olderThanDays` / `retainLast`:** Iceberg
-already defines defaults for both, sourced from table properties:
-
-| Procedure argument | Iceberg fallback (table property) | Iceberg
default |
-| ------------------ | ----------------------------------------- |
--------------- |
-| `older_than` | `history.expire.max-snapshot-age-ms` | 5 days
|
-| `retain_last` | `history.expire.min-snapshots-to-keep` | 1
|
-
-If Gravitino injected its own defaults, a policy that only sets
-`retainLast` would silently also apply age-based expiration, and a
-per-table `history.expire.*` property set by the table owner would be
-overridden by the policy. Omitting unset parameters keeps a single
-condition meaningful on its own and keeps table-level Iceberg
-configuration authoritative.
-
-| Field | Type | Required | Default
| Description |
-| ----------------- | --------- | -------- |
--------------------------------------------- |
------------------------------------------------------------ |
-| `olderThanDays` | `Long` | No | unset - falls back to Iceberg
table property | Expire snapshots older than this many days
|
-| `retainLast` | `Long` | No | unset - falls back to Iceberg
table property | Minimum number of snapshots to always retain
|
-| `streamResults` | `Boolean` | No | `true`
| Stream deleted-file rows to the driver by RDD partition |
-| `minSnapshotCount`| `long` | No | 10
| Trigger threshold - only run once the snapshot count reaches this
|
-
-**Usage scenarios:**
-
-- **Only `olderThanDays`:** Expire snapshots older than N days; the number
- retained follows `history.expire.min-snapshots-to-keep` (Iceberg
- default 1).
-- **Only `retainLast`:** Keep at least N snapshots; the age cutoff follows
- `history.expire.max-snapshot-age-ms` (Iceberg default 5 days).
-- **Both:** Expire snapshots older than N days, but always keep at least M
- snapshots.
-- **Neither:** Pure Iceberg behavior - expiration is governed entirely by
- the table's `history.expire.*` properties.
-
-#### 5.2.4 Why `streamResults` Defaults to `true`
-
-Iceberg's `stream_results` controls how the list of deleted files reaches
-the Spark driver: when `false` (the procedure's own default) the full file
-list is collected to the driver in one result set; when `true` the rows are
-streamed by RDD partition, which is what the Iceberg documentation
-recommends to avoid driver OOM on large deletions.
-
-Setting it to `false` buys very little here: this job consumes only the
-summary counts (`deleted_data_files_count` and friends), never the
-individual file rows, so materializing the whole list on the driver has no
-downstream benefit. The only cost of streaming is slightly more per-
-partition overhead on small tables, which is negligible relative to the
-snapshot expiration itself.
-
-Because the job runs unattended and on tables of unknown size, we default
-`streamResults` to `true` for robustness - deliberately differing from the
-procedure default - while keeping the field configurable for parity with
-the procedure and for operators who want the collected form.
-
-#### 5.2.5 Example Policy Creation
+| Field | Type | Required | Default | Description |
+| --------------- | ------- | -------- | ------- | ----------- |
+| `olderThanDays` | `long` | No | `5` | Age cutoff in days; adapter
emits UTC `older_than` |
+| `retainLast` | `long` | No | `1` | Minimum recent snapshots to
keep |
+| `streamResults` | `boolean`| No | `false` | Forwarded as
`stream_results` |
+
+`olderThanDays` must be in `[1, 36500]`. `retainLast` must be `>= 1`.
+
+Unlike an earlier draft that omitted Gravitino defaults so Iceberg table
+properties alone would apply, the shipped content always materializes
+defaults (aligned with orphan cleanup and with Iceberg's common 5-day / 1
+retain defaults). The adapter always submits explicit `older_than`,
+`retain_last`, and `stream_results`.
+
+#### 5.2.4 Example Policy Creation
```json
POST /metalakes/default/policies
{
"name": "expire_old_snapshots",
- "type": "system_iceberg_snapshot_expiration",
- "comment": "Expire snapshots older than 7 days, keep at least 3",
+ "policyType": "system_iceberg_snapshot_expiration",
"enabled": true,
"content": {
"olderThanDays": 7,
@@ -288,19 +226,7 @@ POST /metalakes/default/policies
}
```
-A user may also specify only one parameter:
-
-```json
-{
- "name": "retain_only_policy",
- "type": "system_iceberg_snapshot_expiration",
- "comment": "Always keep at least 5 snapshots",
- "enabled": true,
- "content": {
- "retainLast": 5
- }
-}
-```
+Empty `content: {}` applies defaults (`5`, `1`, `false`).
---
@@ -309,322 +235,172 @@ A user may also specify only one parameter:
#### 5.3.1 Strategy Handler
```java
-// NEW: maintenance/optimizer/src/main/java/…/handler/snapshot/
-// SnapshotExpirationStrategyHandler.java
-public class SnapshotExpirationStrategyHandler implements StrategyHandler {
-
- public static final String NAME = "iceberg-snapshot-expiration";
-
- private NameIdentifier nameIdentifier;
- private Strategy strategy;
- private Table tableMetadata;
-
- @Override
- public String strategyType() {
- return NAME;
- }
+public class SnapshotExpirationStrategyHandler extends
BaseExpressionStrategyHandler {
+ public static final String NAME =
IcebergSnapshotExpirationContent.STRATEGY_TYPE_VALUE;
@Override
public Set<DataRequirement> dataRequirements() {
- // Table metadata alone is enough: snapshot count comes from the
- // table's reserved properties. No statistics are needed.
- return EnumSet.of(DataRequirement.TABLE_METADATA);
- }
-
- @Override
- public void initialize(StrategyHandlerContext context) {
- Preconditions.checkArgument(
- context.tableMetadata().isPresent(), "Table metadata is null");
- this.nameIdentifier = context.nameIdentifier();
- this.strategy = context.strategy();
- this.tableMetadata = context.tableMetadata().get();
+ return Set.of(DataRequirement.TABLE_METADATA);
}
@Override
- public boolean shouldTrigger() {
- OptionalLong snapshotCount = snapshotCount();
- return snapshotCount.isPresent()
- && snapshotCount.getAsLong() >= minSnapshotCount(strategy);
+ public String strategyType() {
+ return NAME;
}
@Override
- public StrategyEvaluation evaluate() {
- // Score is the snapshot count: tables with more snapshots come first.
- return new StrategyEvaluationImpl(
- snapshotCount().getAsLong(),
- new SnapshotExpirationJobContext(
- nameIdentifier, strategy.jobOptions(),
strategy.jobTemplateName()));
+ protected JobExecutionContext buildJobExecutionContext(...) {
+ return new SnapshotExpirationJobContext(
+ nameIdentifier, jobOptions, strategy.jobTemplateName());
}
-
- // Reads the `snapshot-count` reserved property from table metadata.
- private OptionalLong snapshotCount() { … }
}
```
-**Key design decision 1 - table level, not partition level.** Unlike
-compaction, which scores and selects individual partitions,
-`expire_snapshots` always processes the entire table's snapshot history.
-Therefore no partition scoring or selection logic is needed, and
-`dataRequirements()` excludes `PARTITION_STATISTICS`.
-
-**Key design decision 2 - trigger on table metadata, not statistics.**
-Snapshot count is already part of Iceberg table metadata and needs no scan,
-so the handler reads it from the metadata the Recommender already loads (see
-[§6](#6-snapshot-count-from-table-metadata)) instead of depending on a
-collected statistic. Two consequences:
-
-- The handler implements `StrategyHandler` directly rather than extending
- `BaseExpressionStrategyHandler`. The base class builds its expression
- context from table/partition statistics plus numeric strategy rules only,
- so metadata-derived values are not addressable from `trigger-expr` /
- `score-expr`. With a single table-level condition, the expression
- machinery buys nothing, so the handler compares snapshot count against
- `minSnapshotCount` in `shouldTrigger()` and returns it as the score.
-- Expiration no longer depends on `builtin-iceberg-update-stats` having run
- recently, and it cannot act on a stale snapshot count.
-
-If the `snapshot-count` property is absent (for example, a non-Iceberg table
-or an older catalog), `shouldTrigger()` logs and returns `false` rather than
-guessing.
+**Key design decisions:**
+
+1. **Whole table, not partition-level.** `expire_snapshots` always processes
+ table snapshot history; `dataRequirements` is only `TABLE_METADATA`
+ (same as orphan).
+2. **Orphan-style eligibility.** `trigger-expr` is `true`. Each optimizer
+ invocation may submit when the policy is attached. Retention is enforced
+ by job options (`olderThanDays` / `retainLast`), not by a
+ `snapshot-count ≥ N` gate.
+3. **Register via config:**
+ `gravitino.optimizer.strategyHandler.iceberg-snapshot-expiration.className`.
#### 5.3.2 Job Execution Context
```java
-// NEW: maintenance/optimizer/src/main/java/…/handler/snapshot/
-// SnapshotExpirationJobContext.java
public class SnapshotExpirationJobContext implements JobExecutionContext {
- private final NameIdentifier nameIdentifier;
- private final Map<String, String> jobOptions;
- private final String jobTemplateName;
-
- // jobOptions keys (all optional except catalog/table):
- // older_than_days, retain_last, stream_results,
- // catalog_name, table_identifier
+ // nameIdentifier, jobOptions (olderThanDays, retainLast, streamResults),
+ // jobTemplateName
}
```
-#### 5.3.3 Handler Registration
-
-The strategy handler type `"iceberg-snapshot-expiration"` must be
-registered so that the `Recommender` can instantiate it when it encounters
-a policy with that strategy type. This follows the existing pattern where
-handler classes are looked up by strategy type name.
-
---
### 5.4 Layer 3 - Job Adapter (`maintenance/optimizer/`)
-#### 5.4.1 Job Adapter
-
-The adapter passes only the expiration conditions that were explicitly set
-in the policy content, so each works independently, and always passes
-`stream_results`:
-
```java
-// NEW: maintenance/optimizer/src/main/java/…/job/
-// GravitinoSnapshotExpirationJobAdapter.java
-public class GravitinoSnapshotExpirationJobAdapter
- implements GravitinoJobAdapter {
-
- @Override
- public Map<String, String> jobConfig(JobExecutionContext context) {
- SnapshotExpirationJobContext ctx =
- (SnapshotExpirationJobContext) context;
- Map<String, String> config = new HashMap<>();
- config.put("catalog_name",
- ctx.nameIdentifier().namespace()[0]);
- config.put("table_identifier",
- ctx.nameIdentifier().namespace()[1]
- + "." + ctx.nameIdentifier().name());
-
- Map<String, String> opts = ctx.jobOptions();
-
- // Only pass older_than if explicitly configured
- if (opts.containsKey("olderThanDays")) {
- long days = Long.parseLong(opts.get("olderThanDays"));
- String ts = Instant.now()
- .minus(Duration.ofDays(days))
- .toString()
- .replace("T", " ")
- .substring(0, 19); // "yyyy-MM-dd HH:mm:ss"
- config.put("older_than", ts);
- }
-
- // Only pass retain_last if explicitly configured
- if (opts.containsKey("retainLast")) {
- config.put("retain_last", opts.get("retainLast"));
- }
-
- // stream_results always has a value: policy setting, or `true`
- config.put(
- "stream_results",
- opts.getOrDefault(
- "streamResults",
- String.valueOf(
- IcebergSnapshotExpirationContent.DEFAULT_STREAM_RESULTS)));
-
- return config;
- }
+public class GravitinoSnapshotExpirationJobAdapter implements
GravitinoJobAdapter {
+ // olderThanDays == 5 → omit older_than (Iceberg / table history.expire.*)
+ // olderThanDays != 5 → older_than as UTC timestamp with Z
+ // retainLast == 1 → omit retain_last; else emit retain_last
+ // streamResults → stream_results
+ // plus table_identifier (catalog_name from shared submitter config)
}
```
-#### 5.4.2 Register Adapter
-
-```java
-// UPDATE: GravitinoJobSubmitter.java
-private static final Map<String, Class<? extends GravitinoJobAdapter>>
- jobAdapters = ImmutableMap.of(
- "builtin-iceberg-rewrite-data-files",
- GravitinoCompactionJobAdapter.class,
- "builtin-iceberg-expire-snapshots",
- GravitinoSnapshotExpirationJobAdapter.class // NEW
- );
-```
+Register in `GravitinoJobSubmitter.jobAdapters` under
+`builtin-iceberg-expire-snapshots`.
---
### 5.5 Layer 4 - Spark Job (`maintenance/jobs/`) - Already Implemented
-`IcebergExpireSnapshotsJob` is implemented in PR
-[#11206](https://github.com/apache/gravitino/pull/11206):
-
-- **Template name:** `builtin-iceberg-expire-snapshots`
-- **Version:** `v1`
-- **Parameters:** `--catalog`, `--table`, `--older-than`, `--retain-last`,
- `--stream-results`, `--spark-conf` (all optional except `--catalog` and
- `--table`)
-- **Procedure:**
- `CALL catalog.system.expire_snapshots(table => '…', older_than =>
- TIMESTAMP '…', retain_last => N, stream_results => bool)`
-- **Output:** `deleted_data_files_count`,
- `deleted_manifest_files_count`, `deleted_manifest_lists_count`
-- **Security:** SQL injection prevention via `escapeSqlString()` /
- `escapeSqlIdentifier()`
-- **Tests:** 40 unit tests covering template metadata, argument parsing,
- procedure call building, escaping, and input validation
-
-The job already handles partial parameters gracefully - only parameters
-that are non-empty are included in the procedure call.
+`IcebergExpireSnapshotsJob` / template `builtin-iceberg-expire-snapshots`:
----
+- Parameters: `--catalog`, `--table`, `--older-than`, `--retain-last`,
+ `--stream-results`, `--spark-conf`
+- Procedure: `CALL catalog.system.expire_snapshots(...)`
-## 6. Snapshot Count from Table Metadata
+---
-The trigger input is the table's **snapshot count**. It is taken from table
-metadata rather than from a collected statistic.
+## 6. Trigger Model
-**Why not a statistic.** The alternative would be a `custom-snapshot-count`
-statistic produced by `builtin-iceberg-update-stats` (which today derives
-its statistics from the Iceberg `.files` metadata table). That would make
-expiration depend on the stats job having run recently, and the trigger
-would evaluate a snapshot count that is stale by up to one stats interval -
-for a metric that is free to read from metadata.
+### 6.1 Chosen approach (orphan-aligned)
-**What is available today.** The Recommender loads table metadata through
-`GravitinoTableMetadataProvider`, which returns a Gravitino `Table` over
-REST. For Iceberg tables, `IcebergTable` merges in the reserved properties
-built by `IcebergTablePropertiesUtil.buildReservedProperties(TableMetadata)`
-- currently `format`, `provider`, `current-snapshot-id`, `location`,
-`format-version`, `sort-order`, `identifier-fields`. Snapshot count is not
-among them, so the handler cannot read it yet.
+| Concern | Behavior |
+| ------- | -------- |
+| When to run | Optimizer / external schedule invokes recommender |
+| Policy eligibility | `trigger-expr: true` (always eligible when attached) |
+| What to delete | `older_than` + `retain_last` (+ `stream_results`) |
-**Proposed change.** Expose snapshot count as one more reserved Iceberg
-table property. `TableMetadata.snapshots()` is already in memory when the
-properties are built, so this costs no I/O and no scan:
+This matches common Iceberg / Amoro practice: **periodic expire with a
+retention policy**, not "fire only after snapshot count crosses a
+threshold."
-| Property | Type | Source | Description
|
-| ---------------- | ------ | ----------------------------- |
-------------------------------- |
-| `snapshot-count` | `long` | `TableMetadata.snapshots()` | Number of
snapshots in the table |
+### 6.2 Rejected: `snapshot-count ≥ minSnapshotCount`
-This requires:
+An earlier draft proposed reading a reserved `snapshot-count` table
+property and triggering only when count ≥ `minSnapshotCount` (default 10),
+with a custom `StrategyHandler` instead of `BaseExpressionStrategyHandler`.
-1. A `SNAPSHOT_COUNT` constant in `IcebergConstants` and a reserved entry in
- `IcebergTablePropertiesMetadata` (reserved, like `current-snapshot-id`, so
- users cannot set it).
-2. Populating it in `IcebergTablePropertiesUtil.buildReservedProperties`.
+Rejected because:
-Being reserved and read-only, the property is a safe addition: it shows up
-in `loadTable` responses and is rejected on create/alter.
+- Industry tools (Iceberg maintenance docs, Amoro table-expire) schedule
+ expire and apply age / retain-last; count is not the primary trigger.
+- A count gate adds catalog reserved-property work without matching how
+ operators expect expire to run.
+- Orphan cleanup already established the always-eligible + schedule model
+ for whole-table cleanup.
-An unset or non-numeric `snapshot-count` is treated as "unknown" - the
-handler does not trigger, so a non-Iceberg table attached to the policy is
-skipped rather than mis-evaluated.
+A future optional "skip if fewer than N snapshots" optimization can be
+added without redesigning the policy or handler base class.
---
## 7. File Changes Summary
-### 7.1 New Files
-
-| File
| Layer | Description |
-|
---------------------------------------------------------------------------------
| -------- | ---------------------------------- |
-| `api/…/policy/IcebergSnapshotExpirationContent.java`
| Policy | Policy content with expiration cfg |
-|
`maintenance/optimizer/…/handler/snapshot/SnapshotExpirationStrategyHandler.java`
| Strategy | Trigger / score evaluation |
-| `maintenance/optimizer/…/handler/snapshot/SnapshotExpirationJobContext.java`
| Strategy | Job execution context |
-| `maintenance/optimizer/…/job/GravitinoSnapshotExpirationJobAdapter.java`
| Adapter | Context → job config conversion |
-| `maintenance/jobs/…/iceberg/IcebergExpireSnapshotsJob.java`
| Job | Spark job (already in PR #11206) |
-
-### 7.2 Modified Files
-
-| File | Change
|
-| ------------------------------------------------------- |
-------------------------------------------------------- |
-| `api/…/policy/Policy.java` | Add
`ICEBERG_SNAPSHOT_EXPIRATION` to `BuiltInType` enum |
-| `catalog-common/…/iceberg/IcebergConstants.java` | Add
`SNAPSHOT_COUNT` constant |
-| `catalogs/catalog-lakehouse-iceberg/…/IcebergTablePropertiesMetadata.java` |
Add reserved `snapshot-count` property entry |
-| `catalogs/catalog-lakehouse-iceberg/…/utils/IcebergTablePropertiesUtil.java`
| Populate `snapshot-count` from `TableMetadata.snapshots()` |
-| `maintenance/optimizer/…/job/GravitinoJobSubmitter.java` | Register
expire-snapshots adapter in `jobAdapters` map |
-| `maintenance/jobs/…/BuiltInJobTemplateProvider.java` | Register
`IcebergExpireSnapshotsJob` (in PR #11206) |
-| Handler registry | Register
`SnapshotExpirationStrategyHandler` |
-
-### 7.3 Test Files
-
-| File | Description
|
-| ------------------------------------------------ |
----------------------------------- |
-| `TestIcebergSnapshotExpirationContent.java` | Policy content unit tests
|
-| `TestSnapshotExpirationStrategyHandler.java` | Strategy handler unit
tests, including missing / non-numeric `snapshot-count` |
-| `TestGravitinoSnapshotExpirationJobAdapter.java` | Job adapter unit tests
|
-| `TestIcebergTable.java` (existing) | Assert `snapshot-count`
is exposed and reserved |
-| `TestIcebergExpireSnapshotsJob.java` | Spark job unit tests (in
PR #11206) |
+### 7.1 New / primary files
+
+| File | Layer | Description |
+| ---- | ----- | ----------- |
+| `api/…/policy/IcebergSnapshotExpirationContent.java` | Policy | Content +
`trigger-expr` / `job.options.*` |
+|
`maintenance/optimizer/…/handler/snapshot/SnapshotExpirationStrategyHandler.java`
| Strategy | Orphan-style handler |
+| `maintenance/optimizer/…/handler/snapshot/SnapshotExpirationJobContext.java`
| Strategy | Job context |
+| `maintenance/optimizer/…/job/GravitinoSnapshotExpirationJobAdapter.java` |
Adapter | Options → jobConf |
+| `maintenance/jobs/…/iceberg/IcebergExpireSnapshotsJob.java` | Job | Already
shipped |
+
+### 7.2 Modified files
+
+| File | Change |
+| ---- | ------ |
+| `api/…/policy/Policy.java` / DTO / OpenAPI | Built-in type and REST fields |
+| `maintenance/optimizer/…/job/GravitinoJobSubmitter.java` | Register expire
adapter |
+| Optimizer docs | Handler `className` registration |
+
+### 7.3 Tests
+
+| File | Description |
+| ---- | ----------- |
+| `TestIcebergSnapshotExpirationContent.java` | Policy content |
+| `TestGravitinoSnapshotExpirationJobAdapter.java` | Handler + adapter
(orphan-style) |
+| `TestIcebergExpireSnapshotsJob.java` | Spark job |
---
-## 8. Proposed PR Plan
+## 8. Configuration
-| PR | Scope
| Dependencies |
-| ----------------- |
------------------------------------------------------------------------------------
| ------------ |
-| **PR 1** (this) | Design document
| None |
-| **PR 2** (ready) | Job layer: `IcebergExpireSnapshotsJob` +
`BuiltInJobTemplateProvider` + tests | PR 1 |
-| **PR 3** | Metadata layer: reserved `snapshot-count` Iceberg table
property + tests | PR 1 |
-| **PR 4** | Policy layer: `IcebergSnapshotExpirationContent` +
`Policy.BuiltInType` + tests | PR 2 |
-| **PR 5** | Strategy + Adapter: handler, context, adapter, submitter
registration + tests | PR 3, PR 4 |
+```properties
+gravitino.optimizer.strategyHandler.iceberg-snapshot-expiration.className = \
+
org.apache.gravitino.maintenance.optimizer.recommender.handler.snapshot.SnapshotExpirationStrategyHandler
+```
-PR 3 is independent of PR 2 and can land in parallel. PRs 4 and 5 can be
-combined into a single PR if preferred.
+The job adapter is registered in code. Shared submitter settings still
+supply `catalog_name` and `spark_conf`.
---
## 9. Open Questions
-1. **Trigger conditions** - Should we support conditions beyond snapshot
- count (e.g., oldest snapshot age, total metadata size)? Oldest snapshot
- age would need a second reserved property
- (`oldest-snapshot-timestamp-ms`); it is equally free to compute from
- `TableMetadata`, so it can be added later without redesign.
-2. **PR granularity** - Single PR for all remaining layers or split
- per-layer?
+1. **Optional skip-if-few-snapshots** - Worth a later knob, or leave to
+ Iceberg no-op when nothing is expired?
+2. **Default `streamResults`** - Shipped as `false` (procedure default).
+ Operators may prefer `true` on large tables; field remains configurable.
---
-## 10. Comparison with Compaction Flow
-
-| Aspect | Compaction
| Snapshot Expiration |
-| ----------------- |
-------------------------------------------------------------- |
--------------------------------------- |
-| Policy type | `system_iceberg_compaction`
| `system_iceberg_snapshot_expiration` |
-| Strategy type | `iceberg-data-compaction`
| `iceberg-snapshot-expiration` |
-| Job template | `builtin-iceberg-rewrite-data-files`
| `builtin-iceberg-expire-snapshots` |
-| Scope | Per-partition (scored, top-N selected)
| Whole table |
-| Handler base | `BaseExpressionStrategyHandler` (expression-driven)
| `StrategyHandler` (single metadata check) |
-| Data requirements | `TABLE_METADATA` + `TABLE_STATISTICS` +
`PARTITION_STATISTICS` | `TABLE_METADATA` |
-| Trigger input | `custom-data-file-mse`, `custom-delete-file-number`
(statistics) | `snapshot-count` (table metadata property) |
-| Iceberg procedure | `rewrite_data_files`
| `expire_snapshots` |
-| Key parameters | strategy, sort-order, where, options (all required)
| older_than, retain_last (independently optional, fall back to Iceberg
table properties), stream_results (defaults to `false`) |
+## 10. Comparison with Orphan Cleanup
+
+| Aspect | Orphan cleanup |
Snapshot expiration |
+| ----------------- | --------------------------------------------------- |
-------------------------------------------- |
+| Policy type | `system_iceberg_orphan_file_removal` |
`system_iceberg_snapshot_expiration` |
+| Strategy type | `iceberg-orphan-file-removal` |
`iceberg-snapshot-expiration` |
+| Job template | `builtin-iceberg-remove-orphan-files` |
`builtin-iceberg-expire-snapshots` |
+| Scope | Whole table |
Whole table |
+| Handler base | `BaseExpressionStrategyHandler` |
`BaseExpressionStrategyHandler` |
+| Data requirements | `TABLE_METADATA` |
`TABLE_METADATA` |
+| Trigger | `trigger-expr: true` |
`trigger-expr: true` |
+| Iceberg procedure | `remove_orphan_files` |
`expire_snapshots` |
+| Key parameters | `olderThanDays`, `location`, `dryRun` |
`olderThanDays`, `retainLast`, `streamResults` |
diff --git a/docs/open-api/policies.yaml b/docs/open-api/policies.yaml
index 781080ff47..924eba053b 100644
--- a/docs/open-api/policies.yaml
+++ b/docs/open-api/policies.yaml
@@ -517,8 +517,30 @@ components:
IcebergSnapshotExpirationContent:
type: object
- description: Placeholder content for system_iceberg_snapshot_expiration.
Configurable thresholds land in a follow-up.
+ description: >
+ Snapshot expiration options for system_iceberg_snapshot_expiration.
Scheduling is
+ configured separately. On updateContent, newContent fully replaces the
previous content;
+ omitted fields use these defaults rather than merging with stored
values. Integer fields
+ must be JSON integers (not strings or floats); streamResults must be a
JSON boolean.
additionalProperties: false
+ properties:
+ olderThanDays:
+ type: integer
+ format: int64
+ minimum: 1
+ maximum: 36500
+ default: 5
+ description: Minimum age of snapshots eligible for expiration, in
days (Iceberg default).
+ retainLast:
+ type: integer
+ format: int64
+ minimum: 1
+ default: 1
+ description: Minimum number of recent snapshots to keep regardless
of age.
+ streamResults:
+ type: boolean
+ default: false
+ description: Whether the expire job should stream intermediate
delete results.
IcebergSnapshotExpirationPolicy:
allOf:
diff --git a/docs/table-maintenance-service/optimizer-configuration.md
b/docs/table-maintenance-service/optimizer-configuration.md
index ffa9334358..bd9bd65b46 100644
--- a/docs/table-maintenance-service/optimizer-configuration.md
+++ b/docs/table-maintenance-service/optimizer-configuration.md
@@ -198,9 +198,54 @@ See [Remove Orphan
Files](./optimizer-cli-reference.md#remove-orphan-files) for
complete submission example. For policy-driven submission, configure the
handler
and policy below. Periodic scheduling remains separate.
+### Snapshot expiration policy integration
+
+Register the handler alongside the existing optimizer providers (optional; the
+Recommender also defaults this class when the property is unset):
+
+```properties
+gravitino.optimizer.strategyHandler.iceberg-snapshot-expiration.className =
org.apache.gravitino.maintenance.optimizer.recommender.handler.snapshot.SnapshotExpirationStrategyHandler
+```
+
+The built-in job adapter is registered automatically. Keep the same Spark and
+catalog submission configuration as for direct expire-snapshots runs, including
+`catalog_name` and `spark_conf`. The adapter supplies `table_identifier` and
+`stream_results`. For `older_than` / `retain_last`, values equal to Iceberg's
+procedure defaults (`olderThanDays=5`, `retainLast=1`) are omitted so the job
+template placeholders stay empty and Iceberg table `history.expire.*`
+properties remain authoritative; non-default values are emitted explicitly.
+
+Create a policy through `POST /api/metalakes/{metalake}/policies`:
+
+```json
+{
+ "name": "expire_snapshots",
+ "policyType": "system_iceberg_snapshot_expiration",
+ "enabled": true,
+ "content": {
+ "olderThanDays": 5,
+ "retainLast": 1,
+ "streamResults": false
+ }
+}
+```
+
+Associate the policy with a tag and attach that tag to the target table,
schema,
+or catalog, following the [policy setup walkthrough](./optimizer.md).
+`olderThanDays` defaults to 5 (Iceberg's expire default) and must be between 1
and 36500.
+`retainLast` defaults to 1. `streamResults` defaults to false.
+
+When `olderThanDays` is not the Iceberg default of 5, the adapter subtracts it
+from the current instant and emits `older_than` as an explicit UTC timestamp
+ending in `Z`. Each optimizer invocation may submit expire when the policy is
+attached (`trigger-expr` is `true`). This does not install a periodic
+scheduler, cooldown, or last-run tracking—same as orphan cleanup. Operators
+must control how often `submit-strategy-jobs` (or an external cron) runs.
+
### Orphan cleanup policy integration
-Register the handler alongside the existing optimizer providers:
+Register the handler alongside the existing optimizer providers (optional; the
+Recommender also defaults this class when the property is unset):
```properties
gravitino.optimizer.strategyHandler.iceberg-orphan-file-removal.className =
org.apache.gravitino.maintenance.optimizer.recommender.handler.orphan.OrphanFileRemovalStrategyHandler
diff --git
a/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/Recommender.java
b/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/Recommender.java
index d73ceb4aa0..254836d40a 100644
---
a/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/Recommender.java
+++
b/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/Recommender.java
@@ -50,7 +50,13 @@ import
org.apache.gravitino.maintenance.optimizer.common.OptimizerEnv;
import org.apache.gravitino.maintenance.optimizer.common.conf.OptimizerConfig;
import org.apache.gravitino.maintenance.optimizer.common.util.ProviderUtils;
import
org.apache.gravitino.maintenance.optimizer.recommender.handler.ManifestRewriteStrategyHandler;
+import
org.apache.gravitino.maintenance.optimizer.recommender.handler.compaction.CompactionStrategyHandler;
+import
org.apache.gravitino.maintenance.optimizer.recommender.handler.orphan.OrphanFileRemovalStrategyHandler;
+import
org.apache.gravitino.maintenance.optimizer.recommender.handler.snapshot.SnapshotExpirationStrategyHandler;
+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.rel.Table;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -380,10 +386,22 @@ public class Recommender implements AutoCloseable {
*/
private String getStrategyHandlerClassName(String strategyType) {
String configured =
optimizerEnv.config().getStrategyHandlerClassName(strategyType);
- if (StringUtils.isBlank(configured)
- &&
IcebergRewriteManifestsContent.STRATEGY_TYPE_VALUE.equals(strategyType)) {
+ if (StringUtils.isNotBlank(configured)) {
+ return configured;
+ }
+ // Built-in defaults so policy-driven jobs work without copying docs into
conf.
+ if (IcebergDataCompactionContent.STRATEGY_TYPE_VALUE.equals(strategyType))
{
+ return CompactionStrategyHandler.class.getName();
+ }
+ if
(IcebergRewriteManifestsContent.STRATEGY_TYPE_VALUE.equals(strategyType)) {
return ManifestRewriteStrategyHandler.class.getName();
}
+ if
(IcebergOrphanFileRemovalContent.STRATEGY_TYPE_VALUE.equals(strategyType)) {
+ return OrphanFileRemovalStrategyHandler.class.getName();
+ }
+ if
(IcebergSnapshotExpirationContent.STRATEGY_TYPE_VALUE.equals(strategyType)) {
+ return SnapshotExpirationStrategyHandler.class.getName();
+ }
return configured;
}
diff --git
a/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/snapshot/SnapshotExpirationJobContext.java
b/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/snapshot/SnapshotExpirationJobContext.java
new file mode 100644
index 0000000000..718ee74dd5
--- /dev/null
+++
b/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/snapshot/SnapshotExpirationJobContext.java
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package
org.apache.gravitino.maintenance.optimizer.recommender.handler.snapshot;
+
+import java.util.Map;
+import org.apache.gravitino.NameIdentifier;
+import
org.apache.gravitino.maintenance.optimizer.api.recommender.JobExecutionContext;
+
+/** Immutable target and options for table-level snapshot expiration. */
+public class SnapshotExpirationJobContext implements JobExecutionContext {
+ private final NameIdentifier name;
+ private final Map<String, String> jobOptions;
+ private final String jobTemplateName;
+
+ /**
+ * Creates the context from table metadata and policy options.
+ *
+ * @param name normalized catalog.schema.table identifier
+ * @param jobOptions policy options
+ * @param jobTemplateName template to submit
+ */
+ public SnapshotExpirationJobContext(
+ NameIdentifier name, Map<String, String> jobOptions, String
jobTemplateName) {
+ this.name = name;
+ this.jobOptions = Map.copyOf(jobOptions);
+ this.jobTemplateName = jobTemplateName;
+ }
+
+ @Override
+ public NameIdentifier nameIdentifier() {
+ return name;
+ }
+
+ @Override
+ public Map<String, String> jobOptions() {
+ return jobOptions;
+ }
+
+ @Override
+ public String jobTemplateName() {
+ return jobTemplateName;
+ }
+}
diff --git
a/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/snapshot/SnapshotExpirationStrategyHandler.java
b/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/snapshot/SnapshotExpirationStrategyHandler.java
new file mode 100644
index 0000000000..43c371aa3e
--- /dev/null
+++
b/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/snapshot/SnapshotExpirationStrategyHandler.java
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package
org.apache.gravitino.maintenance.optimizer.recommender.handler.snapshot;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.maintenance.optimizer.api.common.PartitionPath;
+import org.apache.gravitino.maintenance.optimizer.api.common.Strategy;
+import
org.apache.gravitino.maintenance.optimizer.api.recommender.JobExecutionContext;
+import
org.apache.gravitino.maintenance.optimizer.recommender.handler.BaseExpressionStrategyHandler;
+import org.apache.gravitino.policy.IcebergSnapshotExpirationContent;
+import org.apache.gravitino.rel.Table;
+
+/**
+ * Evaluates snapshot expiration at table level, including for partitioned
tables.
+ *
+ * <p>Follows orphan cleanup: {@code trigger-expr} defaults to {@code true},
so each optimizer
+ * invocation may submit expire when the policy is attached. Retention is
enforced by job options
+ * ({@code olderThanDays} / {@code retainLast}), not by a snapshot-count gate.
+ */
+public class SnapshotExpirationStrategyHandler extends
BaseExpressionStrategyHandler {
+
+ /** Registered strategy type. */
+ public static final String NAME =
IcebergSnapshotExpirationContent.STRATEGY_TYPE_VALUE;
+
+ @Override
+ public Set<DataRequirement> dataRequirements() {
+ return Set.of(DataRequirement.TABLE_METADATA);
+ }
+
+ @Override
+ public String strategyType() {
+ return NAME;
+ }
+
+ @Override
+ protected JobExecutionContext buildJobExecutionContext(
+ NameIdentifier nameIdentifier,
+ Strategy strategy,
+ Table tableMetadata,
+ List<PartitionPath> partitions,
+ Map<String, String> jobOptions) {
+ return new SnapshotExpirationJobContext(nameIdentifier, jobOptions,
strategy.jobTemplateName());
+ }
+}
diff --git
a/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/job/GravitinoJobSubmitter.java
b/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/job/GravitinoJobSubmitter.java
index 69fb475611..1a6f62b377 100644
---
a/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/job/GravitinoJobSubmitter.java
+++
b/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/job/GravitinoJobSubmitter.java
@@ -33,6 +33,7 @@ import
org.apache.gravitino.maintenance.optimizer.common.util.GravitinoClientUti
import org.apache.gravitino.policy.IcebergDataCompactionContent;
import org.apache.gravitino.policy.IcebergOrphanFileRemovalContent;
import org.apache.gravitino.policy.IcebergRewriteManifestsContent;
+import org.apache.gravitino.policy.IcebergSnapshotExpirationContent;
/** Submits optimizer jobs to Gravitino using job template adapters. */
public class GravitinoJobSubmitter implements JobSubmitter {
@@ -55,7 +56,9 @@ public class GravitinoJobSubmitter implements JobSubmitter {
IcebergRewriteManifestsContent.JOB_TEMPLATE_NAME_VALUE,
GravitinoManifestRewriteJobAdapter.class,
IcebergOrphanFileRemovalContent.JOB_TEMPLATE_NAME_VALUE,
- GravitinoOrphanFileRemovalJobAdapter.class);
+ GravitinoOrphanFileRemovalJobAdapter.class,
+ IcebergSnapshotExpirationContent.JOB_TEMPLATE_NAME_VALUE,
+ GravitinoSnapshotExpirationJobAdapter.class);
@Override
public String name() {
diff --git
a/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/job/GravitinoSnapshotExpirationJobAdapter.java
b/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/job/GravitinoSnapshotExpirationJobAdapter.java
new file mode 100644
index 0000000000..30d000bf3c
--- /dev/null
+++
b/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/job/GravitinoSnapshotExpirationJobAdapter.java
@@ -0,0 +1,120 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.maintenance.optimizer.recommender.job;
+
+import com.google.common.base.Preconditions;
+import java.time.Clock;
+import java.time.DateTimeException;
+import java.time.ZoneOffset;
+import java.time.format.DateTimeFormatter;
+import java.time.temporal.ChronoUnit;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import
org.apache.gravitino.maintenance.optimizer.api.recommender.JobExecutionContext;
+import org.apache.gravitino.maintenance.optimizer.common.util.IdentifierUtils;
+import
org.apache.gravitino.maintenance.optimizer.recommender.handler.snapshot.SnapshotExpirationJobContext;
+import org.apache.gravitino.policy.IcebergSnapshotExpirationContent;
+
+/** Converts snapshot-expiration options to the built-in Spark template
configuration. */
+public class GravitinoSnapshotExpirationJobAdapter implements
GravitinoJobAdapter {
+ private static final String TABLE_IDENTIFIER_KEY = "table_identifier";
+ private static final String OLDER_THAN_KEY = "older_than";
+ private static final String RETAIN_LAST_KEY = "retain_last";
+ private static final String STREAM_RESULTS_KEY = "stream_results";
+ private static final DateTimeFormatter TIMESTAMP =
+ DateTimeFormatter.ofPattern("uuuu-MM-dd
HH:mm:ssXXX").withZone(ZoneOffset.UTC);
+ private final Clock clock;
+
+ /** Creates an adapter using the UTC system clock. */
+ public GravitinoSnapshotExpirationJobAdapter() {
+ this(Clock.systemUTC());
+ }
+
+ GravitinoSnapshotExpirationJobAdapter(Clock clock) {
+ this.clock = clock;
+ }
+
+ @Override
+ public Map<String, String> jobConfig(JobExecutionContext context) {
+ Preconditions.checkArgument(
+ context instanceof SnapshotExpirationJobContext,
+ "jobExecutionContext must be SnapshotExpirationJobContext");
+ SnapshotExpirationJobContext expire = (SnapshotExpirationJobContext)
context;
+ Map<String, String> options = expire.jobOptions();
+ long days;
+ try {
+ days =
+ Long.parseLong(
+ options.getOrDefault(
+ IcebergSnapshotExpirationContent.OLDER_THAN_DAYS_KEY,
+
String.valueOf(IcebergSnapshotExpirationContent.DEFAULT_OLDER_THAN_DAYS)));
+ } catch (NumberFormatException e) {
+ throw new IllegalArgumentException("olderThanDays must be an integer",
e);
+ }
+ IcebergSnapshotExpirationContent.validateOlderThanDays(days);
+
+ long retainLast;
+ try {
+ retainLast =
+ Long.parseLong(
+ options.getOrDefault(
+ IcebergSnapshotExpirationContent.RETAIN_LAST_KEY,
+
String.valueOf(IcebergSnapshotExpirationContent.DEFAULT_RETAIN_LAST)));
+ } catch (NumberFormatException e) {
+ throw new IllegalArgumentException("retainLast must be an integer", e);
+ }
+ IcebergSnapshotExpirationContent.validateRetainLast(retainLast);
+
+ String streamResults =
+ options.getOrDefault(
+ IcebergSnapshotExpirationContent.STREAM_RESULTS_KEY,
+
String.valueOf(IcebergSnapshotExpirationContent.DEFAULT_STREAM_RESULTS));
+ Preconditions.checkArgument(
+ "true".equalsIgnoreCase(streamResults) ||
"false".equalsIgnoreCase(streamResults),
+ "streamResults must be true or false");
+
+ Map<String, String> config = new LinkedHashMap<>();
+ config.put(
+ TABLE_IDENTIFIER_KEY,
+
IdentifierUtils.removeCatalogFromIdentifier(expire.nameIdentifier()).toString());
+ // Match template placeholders {{older_than:-}} / {{retain_last:-}}: omit
when using the same
+ // defaults as Iceberg's procedure so table history.expire.* properties
remain authoritative.
+ if (days == IcebergSnapshotExpirationContent.DEFAULT_OLDER_THAN_DAYS) {
+ config.put(OLDER_THAN_KEY, "");
+ } else {
+ config.put(OLDER_THAN_KEY, cutoff(days));
+ }
+ if (retainLast == IcebergSnapshotExpirationContent.DEFAULT_RETAIN_LAST) {
+ config.put(RETAIN_LAST_KEY, "");
+ } else {
+ config.put(RETAIN_LAST_KEY, Long.toString(retainLast));
+ }
+ config.put(STREAM_RESULTS_KEY,
Boolean.toString(Boolean.parseBoolean(streamResults)));
+ return Map.copyOf(config);
+ }
+
+ private String cutoff(long days) {
+ try {
+ return TIMESTAMP.format(clock.instant().minus(days, ChronoUnit.DAYS));
+ } catch (DateTimeException | ArithmeticException e) {
+ throw new IllegalArgumentException(
+ "olderThanDays cannot be represented as an expiration timestamp", e);
+ }
+ }
+}
diff --git
a/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/recommender/job/TestGravitinoSnapshotExpirationJobAdapter.java
b/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/recommender/job/TestGravitinoSnapshotExpirationJobAdapter.java
new file mode 100644
index 0000000000..3f8a6f896f
--- /dev/null
+++
b/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/recommender/job/TestGravitinoSnapshotExpirationJobAdapter.java
@@ -0,0 +1,149 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.maintenance.optimizer.recommender.job;
+
+import java.time.Clock;
+import java.time.Instant;
+import java.time.ZoneOffset;
+import java.util.Map;
+import java.util.Set;
+import org.apache.gravitino.NameIdentifier;
+import
org.apache.gravitino.maintenance.optimizer.api.recommender.StrategyHandler.DataRequirement;
+import
org.apache.gravitino.maintenance.optimizer.api.recommender.StrategyHandlerContext;
+import org.apache.gravitino.maintenance.optimizer.common.conf.OptimizerConfig;
+import
org.apache.gravitino.maintenance.optimizer.recommender.handler.snapshot.SnapshotExpirationJobContext;
+import
org.apache.gravitino.maintenance.optimizer.recommender.handler.snapshot.SnapshotExpirationStrategyHandler;
+import
org.apache.gravitino.maintenance.optimizer.recommender.strategy.GravitinoStrategy;
+import org.apache.gravitino.policy.IcebergSnapshotExpirationContent;
+import org.apache.gravitino.policy.Policy;
+import org.apache.gravitino.policy.PolicyContents;
+import org.apache.gravitino.rel.Table;
+import org.apache.gravitino.rel.expressions.transforms.Transform;
+import org.apache.gravitino.rel.expressions.transforms.Transforms;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+class TestGravitinoSnapshotExpirationJobAdapter {
+ private static final String TEMPLATE =
IcebergSnapshotExpirationContent.JOB_TEMPLATE_NAME_VALUE;
+ private static final Clock CLOCK =
+ Clock.fixed(Instant.parse("2026-09-27T12:30:00Z"), ZoneOffset.UTC);
+ private final GravitinoSnapshotExpirationJobAdapter adapter =
+ new GravitinoSnapshotExpirationJobAdapter(CLOCK);
+
+ @Test
+ void policyToSubmissionWorksWithoutPartitionStatistics() {
+ for (boolean partitioned : new boolean[] {false, true}) {
+ Policy policy = Mockito.mock(Policy.class);
+ Mockito.when(policy.content())
+ .thenReturn(PolicyContents.icebergSnapshotExpiration(5L, 3L, true));
+ GravitinoStrategy strategy = new GravitinoStrategy(policy);
+ Table table = Mockito.mock(Table.class);
+ Mockito.when(table.properties()).thenReturn(Map.of());
+ Mockito.when(table.partitioning())
+ .thenReturn(partitioned ? new Transform[]
{Transforms.identity("id")} : new Transform[0]);
+ SnapshotExpirationStrategyHandler handler = new
SnapshotExpirationStrategyHandler();
+ handler.initialize(
+ StrategyHandlerContext.builder(NameIdentifier.of("catalog", "db",
"table"), strategy)
+ .withTableMetadata(table)
+ .build());
+ Assertions.assertEquals(Set.of(DataRequirement.TABLE_METADATA),
handler.dataRequirements());
+ Assertions.assertEquals(strategy.strategyType(), handler.strategyType());
+ Assertions.assertTrue(handler.shouldTrigger());
+ Assertions.assertEquals(1, handler.evaluate().score());
+ SnapshotExpirationJobContext context =
+ (SnapshotExpirationJobContext)
handler.evaluate().jobExecutionContext().orElseThrow();
+ Assertions.assertInstanceOf(
+ GravitinoSnapshotExpirationJobAdapter.class,
+ new
GravitinoJobSubmitter().loadJobAdapter(context.jobTemplateName()));
+ OptimizerConfig config =
+ new OptimizerConfig(
+ Map.of(
+ OptimizerConfig.JOB_SUBMITTER_CONFIG_PREFIX +
"catalog_name", "spark_catalog"));
+ Map<String, String> job = GravitinoJobSubmitter.buildJobConfig(config,
context, adapter);
+ Assertions.assertEquals("spark_catalog", job.get("catalog_name"));
+ Assertions.assertEquals("db.table", job.get("table_identifier"));
+ // olderThanDays=5 matches Iceberg default → omit so table properties
can apply
+ Assertions.assertEquals("", job.get("older_than"));
+ Assertions.assertEquals("3", job.get("retain_last"));
+ Assertions.assertEquals("true", job.get("stream_results"));
+ }
+ }
+
+ @Test
+ void defaultsFillOptionalPlaceholders() {
+ Map<String, String> job = adapter.jobConfig(context(Map.of()));
+ Assertions.assertEquals("", job.get("older_than"));
+ Assertions.assertEquals("", job.get("retain_last"));
+ Assertions.assertEquals("false", job.get("stream_results"));
+ Assertions.assertEquals(
+ "2026-09-26 12:30:00Z",
+ adapter.jobConfig(context(Map.of("olderThanDays",
"1"))).get("older_than"));
+ }
+
+ @Test
+ void rejectsUnsafeOptionsBeforeSubmission() {
+ for (String days : new String[] {"0", "-1", "no", "9223372036854775807"}) {
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> adapter.jobConfig(context(Map.of("olderThanDays", days))));
+ }
+ for (String retain : new String[] {"0", "-1", "no"}) {
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> adapter.jobConfig(context(Map.of("retainLast", retain))));
+ }
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> adapter.jobConfig(context(Map.of("streamResults", "yes"))));
+ }
+
+ @Test
+ void validatesRetentionRangeAndReportsTimestampErrors() {
+ long maximum = IcebergSnapshotExpirationContent.MAX_OLDER_THAN_DAYS;
+ Assertions.assertEquals(
+ "1926-10-22 12:30:00Z",
+ adapter
+ .jobConfig(context(Map.of("olderThanDays",
String.valueOf(maximum))))
+ .get("older_than"));
+ for (String days :
+ new String[] {
+ "0", "-1", "no", String.valueOf(maximum + 1),
String.valueOf(Long.MAX_VALUE)
+ }) {
+ IllegalArgumentException error =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> adapter.jobConfig(context(Map.of("olderThanDays", days))));
+ Assertions.assertTrue(error.getMessage().contains("olderThanDays"));
+ }
+ GravitinoSnapshotExpirationJobAdapter ancient =
+ new GravitinoSnapshotExpirationJobAdapter(Clock.fixed(Instant.MIN,
ZoneOffset.UTC));
+ IllegalArgumentException error =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> ancient.jobConfig(context(Map.of("olderThanDays", "2"))));
+ Assertions.assertTrue(error.getMessage().contains("olderThanDays"));
+ Assertions.assertNotNull(error.getCause());
+ }
+
+ private SnapshotExpirationJobContext context(Map<String, String> options) {
+ return new SnapshotExpirationJobContext(
+ NameIdentifier.of("catalog", "db", "table"), options, TEMPLATE);
+ }
+}
