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 65f44cd106 [#11195] feat(maintenance): Integrate orphan cleanup
policies with the optimizer (#13531)
65f44cd106 is described below
commit 65f44cd10667f24bcf12de315be6bd053ba7b2fa
Author: Akshay Thorat <[email protected]>
AuthorDate: Thu Oct 8 01:02:52 2026 -0700
[#11195] feat(maintenance): Integrate orphan cleanup policies with the
optimizer (#13531)
### What changes were proposed in this pull request?
Add `system_iceberg_orphan_file_removal` policy content, REST
serialization, a table-level strategy handler, and an adapter for the
existing cleanup job from #13153. Enabled policies selected through tags
can now submit orphan cleanup jobs through the optimizer, including for
partitioned tables without partition statistics.
Convert retention days to an explicit UTC cutoff and validate custom
scan locations before submission. Reuse the job's lexical containment
checks; filesystem validation remains in the Spark job. Document policy
configuration and add the Iceberg catalog build prerequisites for
optimizer integration tests.
### Why are the changes needed?
#13153 provided direct job submission. This completes the
policy-to-strategy-to-adapter integration specified in #11700.
Fix: #11195
### Does this PR introduce _any_ user-facing change?
Yes. The new policy accepts `olderThanDays` (default 3, range 1..36500),
optional `location`, and `dryRun` (default false). Register the orphan
strategy handler in optimizer configuration. Every explicit
`submit-strategy-jobs` invocation for an enabled, selected orphan policy
submits a Spark cleanup job. The trigger is always true and the score is
1; there is no metric gate, cooldown, or last-run check. Operators must
rate-limit invocations through external scheduling and should start with
`dryRun: true` to inspect candidates. Dry-run jobs also consume Spark
resources and need rate limiting.
### How was this patch tested?
With JDK 17:
- API, common, optimizer API, optimizer, and jobs module `check` tasks
passed: 1,174 tests passed and 6 skipped.
- REST/tag integration passed for a partitioned table, including
disabled-policy rejection and no submission.
- Spark tests passed, including cutoff handling under a non-UTC session
timezone and direct execution with blank locations (dry-run and
deletion).
- OpenAPI validation (`:docs:build`), Javadoc generation, `rat`, and
diff whitespace checks passed. Javadocs report existing warnings in
unchanged code.
---
.../policy/IcebergOrphanFileRemovalContent.java | 163 +++++++++++++++++++
.../java/org/apache/gravitino/policy/Policy.java | 10 +-
.../apache/gravitino/policy/PolicyContents.java | 24 +++
.../TestIcebergOrphanFileRemovalContent.java | 97 ++++++++++++
.../gravitino/dto/policy/PolicyContentDTO.java | 69 ++++++++
.../org/apache/gravitino/dto/policy/PolicyDTO.java | 5 +-
.../dto/requests/PolicyCreateRequest.java | 5 +-
.../dto/requests/PolicyUpdateRequest.java | 5 +-
.../apache/gravitino/dto/util/DTOConverters.java | 15 ++
.../dto/policy/TestOrphanFileRemovalPolicyDTO.java | 122 ++++++++++++++
docs/open-api/policies.yaml | 176 +++++++++++++++++++--
.../optimizer-configuration.md | 64 +++++++-
docs/table-maintenance-service/optimizer.md | 13 +-
.../jobs/iceberg/IcebergRemoveOrphanFilesJob.java | 44 +-----
.../TestIcebergRemoveOrphanFilesJobWithSpark.java | 43 +++++
.../common/util/OrphanFileLocationUtils.java | 79 +++++++++
maintenance/optimizer/build.gradle.kts | 3 +
.../handler/BaseExpressionStrategyHandler.java | 5 +-
.../orphan/OrphanFileRemovalJobContext.java | 74 +++++++++
.../orphan/OrphanFileRemovalStrategyHandler.java | 65 ++++++++
.../recommender/job/GravitinoJobSubmitter.java | 5 +-
.../job/GravitinoOrphanFileRemovalJobAdapter.java | 104 ++++++++++++
.../optimizer/integration/test/RecommenderIT.java | 169 ++++++++++++++++++++
.../recommender/job/TestGravitinoJobSubmitter.java | 9 ++
.../TestGravitinoOrphanFileRemovalJobAdapter.java | 159 +++++++++++++++++++
25 files changed, 1462 insertions(+), 65 deletions(-)
diff --git
a/api/src/main/java/org/apache/gravitino/policy/IcebergOrphanFileRemovalContent.java
b/api/src/main/java/org/apache/gravitino/policy/IcebergOrphanFileRemovalContent.java
new file mode 100644
index 0000000000..2823901b7b
--- /dev/null
+++
b/api/src/main/java/org/apache/gravitino/policy/IcebergOrphanFileRemovalContent.java
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.policy;
+
+import com.google.common.base.Preconditions;
+import 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.commons.lang3.StringUtils;
+import org.apache.gravitino.MetadataObject;
+
+/** Configuration for table-level Iceberg orphan cleanup on an optimizer
invocation. */
+public class IcebergOrphanFileRemovalContent implements PolicyContent {
+ /** Property key for strategy type. */
+ public static final String STRATEGY_TYPE_KEY = "strategy.type";
+ /** Property key for the job template. */
+ public static final String JOB_TEMPLATE_NAME_KEY = "job.template-name";
+ /** Rule key for the trigger expression. */
+ 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 option key shared with the job adapter. */
+ public static final String OLDER_THAN_DAYS_KEY = "olderThanDays";
+ /** Scan location option key shared with the job adapter. */
+ public static final String LOCATION_KEY = "location";
+ /** Preview mode option key shared with the job adapter. */
+ public static final String DRY_RUN_KEY = "dryRun";
+ /** Maximum retention, approximately 100 years, to keep cleanup timestamps
practical. */
+ public static final long MAX_OLDER_THAN_DAYS = 36500;
+ /** Strategy type for orphan cleanup. */
+ public static final String STRATEGY_TYPE_VALUE =
"iceberg-orphan-file-removal";
+ /** Existing Spark job template used for cleanup. */
+ public static final String JOB_TEMPLATE_NAME_VALUE =
"builtin-iceberg-remove-orphan-files";
+ /** Default retention in days. */
+ public static final long DEFAULT_OLDER_THAN_DAYS = 3;
+ /** Default execution mode. */
+ public static final boolean DEFAULT_DRY_RUN = false;
+
+ private final long olderThanDays;
+ @Nullable private final String location;
+ private final boolean dryRun;
+
+ private IcebergOrphanFileRemovalContent() {
+ this(DEFAULT_OLDER_THAN_DAYS, null, DEFAULT_DRY_RUN);
+ }
+
+ IcebergOrphanFileRemovalContent(long olderThanDays, @Nullable String
location, boolean dryRun) {
+ this.olderThanDays = olderThanDays;
+ this.location = location;
+ this.dryRun = dryRun;
+ }
+
+ /**
+ * @return minimum age of files eligible for removal, in days
+ */
+ public long olderThanDays() {
+ return olderThanDays;
+ }
+
+ /**
+ * @return optional scan location within the table, or null for the table
location
+ */
+ @Nullable
+ public String location() {
+ return location;
+ }
+
+ /**
+ * @return whether candidates should only be listed
+ */
+ public boolean dryRun() {
+ return dryRun;
+ }
+
+ @Override
+ public Set<MetadataObject.Type> supportedObjectTypes() {
+ return ImmutableSet.of(
+ MetadataObject.Type.CATALOG, MetadataObject.Type.SCHEMA,
MetadataObject.Type.TABLE);
+ }
+
+ @Override
+ public Map<String, String> properties() {
+ return ImmutableMap.of(
+ STRATEGY_TYPE_KEY, STRATEGY_TYPE_VALUE, JOB_TEMPLATE_NAME_KEY,
JOB_TEMPLATE_NAME_VALUE);
+ }
+
+ @Override
+ public Map<String, Object> rules() {
+ Map<String, Object> rules = new LinkedHashMap<>();
+ // 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 + DRY_RUN_KEY, dryRun);
+ if (location != null) {
+ rules.put(JOB_OPTIONS_PREFIX + LOCATION_KEY, location);
+ }
+ return Collections.unmodifiableMap(rules);
+ }
+
+ /**
+ * Validates retention before storing a policy or submitting a job.
+ *
+ * @param days minimum file 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);
+ }
+
+ @Override
+ public void validate() {
+ PolicyContent.super.validate();
+ validateOlderThanDays(olderThanDays);
+ Preconditions.checkArgument(
+ location == null || StringUtils.isNotBlank(location), "location must
not be blank");
+ Preconditions.checkArgument(
+ location == null ||
location.equals(StringUtils.strip(location.trim())),
+ "location must not have leading or trailing whitespace");
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ if (!(other instanceof IcebergOrphanFileRemovalContent)) {
+ return false;
+ }
+ IcebergOrphanFileRemovalContent that = (IcebergOrphanFileRemovalContent)
other;
+ return olderThanDays == that.olderThanDays
+ && dryRun == that.dryRun
+ && Objects.equals(location, that.location);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(olderThanDays, location, dryRun);
+ }
+}
diff --git a/api/src/main/java/org/apache/gravitino/policy/Policy.java
b/api/src/main/java/org/apache/gravitino/policy/Policy.java
index d7173d233f..04d30b45c1 100644
--- a/api/src/main/java/org/apache/gravitino/policy/Policy.java
+++ b/api/src/main/java/org/apache/gravitino/policy/Policy.java
@@ -42,10 +42,12 @@ public interface Policy extends Auditable {
ICEBERG_COMPACTION(
BUILT_IN_TYPE_PREFIX + "iceberg_compaction",
IcebergDataCompactionContent.class),
- /**
- * Custom policy type. "custom" is a fixed string that indicates the
policy is a non-built-in
- * type.
- */
+ /** Iceberg orphan file cleanup policy. */
+ ICEBERG_ORPHAN_FILE_REMOVAL(
+ BUILT_IN_TYPE_PREFIX + "iceberg_orphan_file_removal",
+ IcebergOrphanFileRemovalContent.class),
+
+ /** Non-built-in policies use the fixed wire value {@code custom}. */
CUSTOM("custom", PolicyContents.CustomContent.class);
private final String policyType;
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 4b6acc9b94..f767d98f75 100644
--- a/api/src/main/java/org/apache/gravitino/policy/PolicyContents.java
+++ b/api/src/main/java/org/apache/gravitino/policy/PolicyContents.java
@@ -22,6 +22,7 @@ import com.google.common.collect.ImmutableSet;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
+import javax.annotation.Nullable;
import org.apache.gravitino.MetadataObject;
/** Utility class for creating instances of {@link PolicyContent}. */
@@ -103,6 +104,29 @@ public class PolicyContents {
rewriteOptions);
}
+ /**
+ * @return orphan cleanup policy content with the default retention and
execution mode
+ */
+ public static PolicyContent icebergOrphanFileRemoval() {
+ return icebergOrphanFileRemoval(
+ IcebergOrphanFileRemovalContent.DEFAULT_OLDER_THAN_DAYS,
+ null,
+ IcebergOrphanFileRemovalContent.DEFAULT_DRY_RUN);
+ }
+
+ /**
+ * Creates orphan cleanup policy content.
+ *
+ * @param olderThanDays minimum file age in days, from 1 to 36500 inclusive
+ * @param location optional scan location within the table
+ * @param dryRun whether to list candidates without deleting them
+ * @return orphan cleanup policy content
+ */
+ public static PolicyContent icebergOrphanFileRemoval(
+ long olderThanDays, @Nullable String location, boolean dryRun) {
+ return new IcebergOrphanFileRemovalContent(olderThanDays, location,
dryRun);
+ }
+
private PolicyContents() {}
/**
diff --git
a/api/src/test/java/org/apache/gravitino/policy/TestIcebergOrphanFileRemovalContent.java
b/api/src/test/java/org/apache/gravitino/policy/TestIcebergOrphanFileRemovalContent.java
new file mode 100644
index 0000000000..f082ecaad9
--- /dev/null
+++
b/api/src/test/java/org/apache/gravitino/policy/TestIcebergOrphanFileRemovalContent.java
@@ -0,0 +1,97 @@
+/*
+ * 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 org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class TestIcebergOrphanFileRemovalContent {
+ @Test
+ void defaultsAndIdentity() {
+ IcebergOrphanFileRemovalContent content =
+ (IcebergOrphanFileRemovalContent)
PolicyContents.icebergOrphanFileRemoval();
+ content.validate();
+ Assertions.assertEquals(3, content.olderThanDays());
+ Assertions.assertFalse(content.dryRun());
+ Assertions.assertNull(content.location());
+ Assertions.assertEquals(content,
PolicyContents.icebergOrphanFileRemoval(3, null, false));
+ Assertions.assertEquals(
+ content.hashCode(),
PolicyContents.icebergOrphanFileRemoval().hashCode());
+ Assertions.assertNotEquals(content,
PolicyContents.icebergOrphanFileRemoval(4, null, false));
+ Assertions.assertEquals(
+ Policy.BuiltInType.ICEBERG_ORPHAN_FILE_REMOVAL,
+
Policy.BuiltInType.fromPolicyType("system_iceberg_orphan_file_removal"));
+ Assertions.assertEquals(
+ IcebergOrphanFileRemovalContent.class,
+ Policy.BuiltInType.ICEBERG_ORPHAN_FILE_REMOVAL.contentClass());
+ Assertions.assertThrows(
+ UnsupportedOperationException.class, () ->
content.rules().put("job.options.dryRun", true));
+ }
+
+ @Test
+ void validatesRetentionAndLocation() {
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> PolicyContents.icebergOrphanFileRemoval(0, null,
true).validate());
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> PolicyContents.icebergOrphanFileRemoval(-1, null,
false).validate());
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> PolicyContents.icebergOrphanFileRemoval(1, " ",
true).validate());
+ PolicyContent content =
+ PolicyContents.icebergOrphanFileRemoval(1, "s3://bucket/table/data",
true);
+ content.validate();
+ Assertions.assertEquals("s3://bucket/table/data",
content.rules().get("job.options.location"));
+ }
+
+ @Test
+ void rejectsLocationWhitespace() {
+ for (String location :
+ new String[] {
+ " s3://bucket/table/data",
+ "s3://bucket/table/data ",
+ "\ts3://bucket/table/data",
+ "s3://bucket/table/data\n",
+ "\u2003s3://bucket/table/data",
+ "s3://bucket/table/data\u2003",
+ "\u2003"
+ }) {
+ IllegalArgumentException error =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> PolicyContents.icebergOrphanFileRemoval(3, location,
true).validate());
+ Assertions.assertTrue(error.getMessage().contains("location"));
+ }
+ }
+
+ @Test
+ void validatesRetentionUpperBoundary() {
+ long maximum = IcebergOrphanFileRemovalContent.MAX_OLDER_THAN_DAYS;
+ Assertions.assertDoesNotThrow(
+ () -> PolicyContents.icebergOrphanFileRemoval(maximum, null,
true).validate());
+ for (long days : new long[] {maximum + 1, Long.MAX_VALUE}) {
+ IllegalArgumentException error =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> PolicyContents.icebergOrphanFileRemoval(days, null,
true).validate());
+ Assertions.assertTrue(error.getMessage().contains("olderThanDays"));
+ }
+ }
+}
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 5bde7f9af1..a0c2b71616 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
@@ -23,12 +23,14 @@ import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Set;
+import javax.annotation.Nullable;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.EqualsAndHashCode;
import lombok.ToString;
import org.apache.gravitino.MetadataObject;
import org.apache.gravitino.policy.IcebergDataCompactionContent;
+import org.apache.gravitino.policy.IcebergOrphanFileRemovalContent;
import org.apache.gravitino.policy.PolicyContent;
import org.apache.gravitino.policy.PolicyContents;
@@ -204,4 +206,71 @@ public interface PolicyContentDTO extends PolicyContent {
rewriteOptions());
}
}
+
+ /** Typed orphan cleanup policy content for REST requests and responses. */
+ @EqualsAndHashCode
+ @ToString
+ @Builder(setterPrefix = "with")
+ @AllArgsConstructor(access = lombok.AccessLevel.PRIVATE)
+ class IcebergOrphanFileRemovalContentDTO implements PolicyContentDTO {
+ @JsonProperty("olderThanDays")
+ private Long olderThanDays;
+
+ @JsonProperty("location")
+ private String location;
+
+ @JsonProperty("dryRun")
+ private Boolean dryRun;
+
+ private IcebergOrphanFileRemovalContentDTO() {}
+
+ /**
+ * @return minimum age in days, defaulting to three
+ */
+ public long olderThanDays() {
+ return olderThanDays == null
+ ? IcebergOrphanFileRemovalContent.DEFAULT_OLDER_THAN_DAYS
+ : olderThanDays;
+ }
+
+ /**
+ * @return optional scan location
+ */
+ @Nullable
+ public String location() {
+ return location;
+ }
+
+ /**
+ * @return whether to list candidates without deleting them
+ */
+ public boolean dryRun() {
+ return dryRun == null ? IcebergOrphanFileRemovalContent.DEFAULT_DRY_RUN
: dryRun;
+ }
+
+ @Override
+ public Set<MetadataObject.Type> supportedObjectTypes() {
+ return toDomainContent().supportedObjectTypes();
+ }
+
+ @Override
+ public Map<String, String> properties() {
+ return toDomainContent().properties();
+ }
+
+ @Override
+ public Map<String, Object> rules() {
+ return toDomainContent().rules();
+ }
+
+ @Override
+ public void validate() {
+ PolicyContentDTO.super.validate();
+ toDomainContent().validate();
+ }
+
+ private PolicyContent toDomainContent() {
+ return PolicyContents.icebergOrphanFileRemoval(olderThanDays(),
location(), dryRun());
+ }
+ }
}
diff --git
a/common/src/main/java/org/apache/gravitino/dto/policy/PolicyDTO.java
b/common/src/main/java/org/apache/gravitino/dto/policy/PolicyDTO.java
index 51c1780879..ce65479c25 100644
--- a/common/src/main/java/org/apache/gravitino/dto/policy/PolicyDTO.java
+++ b/common/src/main/java/org/apache/gravitino/dto/policy/PolicyDTO.java
@@ -57,7 +57,10 @@ public class PolicyDTO implements Policy {
// "system_data_compaction")
@JsonSubTypes.Type(
value = PolicyContentDTO.IcebergCompactionContentDTO.class,
- name = "system_iceberg_compaction")
+ name = "system_iceberg_compaction"),
+ @JsonSubTypes.Type(
+ value = PolicyContentDTO.IcebergOrphanFileRemovalContentDTO.class,
+ name = "system_iceberg_orphan_file_removal")
})
private PolicyContentDTO content;
diff --git
a/common/src/main/java/org/apache/gravitino/dto/requests/PolicyCreateRequest.java
b/common/src/main/java/org/apache/gravitino/dto/requests/PolicyCreateRequest.java
index 158bbb3e43..9ec2e0ccf9 100644
---
a/common/src/main/java/org/apache/gravitino/dto/requests/PolicyCreateRequest.java
+++
b/common/src/main/java/org/apache/gravitino/dto/requests/PolicyCreateRequest.java
@@ -61,7 +61,10 @@ public class PolicyCreateRequest implements RESTRequest {
// "system_data_compaction")
@JsonSubTypes.Type(
value = PolicyContentDTO.IcebergCompactionContentDTO.class,
- name = "system_iceberg_compaction")
+ name = "system_iceberg_compaction"),
+ @JsonSubTypes.Type(
+ value = PolicyContentDTO.IcebergOrphanFileRemovalContentDTO.class,
+ name = "system_iceberg_orphan_file_removal")
})
private final PolicyContentDTO policyContent;
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 2cc53c68d0..4e362466be 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
@@ -142,7 +142,10 @@ public interface PolicyUpdateRequest extends RESTRequest {
// "system_data_compaction")
@JsonSubTypes.Type(
value = PolicyContentDTO.IcebergCompactionContentDTO.class,
- name = "system_iceberg_compaction")
+ name = "system_iceberg_compaction"),
+ @JsonSubTypes.Type(
+ value = PolicyContentDTO.IcebergOrphanFileRemovalContentDTO.class,
+ name = "system_iceberg_orphan_file_removal")
})
private final PolicyContentDTO newContent;
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 6d3c4881f2..14f6bc6d64 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
@@ -105,6 +105,7 @@ import org.apache.gravitino.messaging.Topic;
import org.apache.gravitino.model.Model;
import org.apache.gravitino.model.ModelVersion;
import org.apache.gravitino.policy.IcebergDataCompactionContent;
+import org.apache.gravitino.policy.IcebergOrphanFileRemovalContent;
import org.apache.gravitino.policy.PolicyContent;
import org.apache.gravitino.policy.PolicyContents;
import org.apache.gravitino.rel.Column;
@@ -675,6 +676,14 @@ public class DTOConverters {
.build();
}
+ if (policyContent instanceof IcebergOrphanFileRemovalContent) {
+ IcebergOrphanFileRemovalContent content =
(IcebergOrphanFileRemovalContent) policyContent;
+ return PolicyContentDTO.IcebergOrphanFileRemovalContentDTO.builder()
+ .withOlderThanDays(content.olderThanDays())
+ .withLocation(content.location())
+ .withDryRun(content.dryRun())
+ .build();
+ }
if (policyContent instanceof IcebergDataCompactionContent) {
IcebergDataCompactionContent icebergCompactionContent =
(IcebergDataCompactionContent) policyContent;
@@ -1532,6 +1541,12 @@ public class DTOConverters {
customContentDTO.properties());
}
+ if (policyContentDTO instanceof
PolicyContentDTO.IcebergOrphanFileRemovalContentDTO) {
+ PolicyContentDTO.IcebergOrphanFileRemovalContentDTO content =
+ (PolicyContentDTO.IcebergOrphanFileRemovalContentDTO)
policyContentDTO;
+ return PolicyContents.icebergOrphanFileRemoval(
+ content.olderThanDays(), content.location(), content.dryRun());
+ }
if (policyContentDTO instanceof
PolicyContentDTO.IcebergCompactionContentDTO) {
PolicyContentDTO.IcebergCompactionContentDTO icebergCompactionContentDTO
=
(PolicyContentDTO.IcebergCompactionContentDTO) policyContentDTO;
diff --git
a/common/src/test/java/org/apache/gravitino/dto/policy/TestOrphanFileRemovalPolicyDTO.java
b/common/src/test/java/org/apache/gravitino/dto/policy/TestOrphanFileRemovalPolicyDTO.java
new file mode 100644
index 0000000000..567270c553
--- /dev/null
+++
b/common/src/test/java/org/apache/gravitino/dto/policy/TestOrphanFileRemovalPolicyDTO.java
@@ -0,0 +1,122 @@
+/*
+ * 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 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.IcebergOrphanFileRemovalContent;
+import org.apache.gravitino.policy.PolicyContent;
+import org.apache.gravitino.policy.PolicyContents;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class TestOrphanFileRemovalPolicyDTO {
+ @Test
+ void createRequestDefaultsAndValidation() throws Exception {
+ String json =
+
"{\"name\":\"cleanup\",\"policyType\":\"system_iceberg_orphan_file_removal\",\"enabled\":true,\"content\":{}}";
+ PolicyCreateRequest request =
+ JsonUtils.objectMapper().readValue(json, PolicyCreateRequest.class);
+ request.validate();
+ Assertions.assertEquals(
+ PolicyContents.icebergOrphanFileRemoval(),
+ 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_orphan_file_removal\",\"newContent\":{\"olderThanDays\":7,\"dryRun\":true}}";
+ PolicyUpdateRequest request =
+ JsonUtils.objectMapper().readValue(json, PolicyUpdateRequest.class);
+ request.validate();
+ PolicyUpdateRequest.UpdatePolicyContentRequest update =
+ (PolicyUpdateRequest.UpdatePolicyContentRequest) request;
+ Assertions.assertEquals(
+ PolicyContents.icebergOrphanFileRemoval(7, null, true),
+ DTOConverters.fromDTO(update.getNewContent()));
+ Assertions.assertEquals(
+ request,
+ JsonUtils.objectMapper()
+ .readValue(
+ JsonUtils.objectMapper().writeValueAsString(request),
PolicyUpdateRequest.class));
+ }
+
+ @Test
+ void responseRoundTripRetainsTypedContent() throws Exception {
+ PolicyContent domain =
+ PolicyContents.icebergOrphanFileRemoval(7, "s3://bucket/table/data",
true);
+ PolicyDTO dto =
+ PolicyDTO.builder()
+ .withName("cleanup")
+ .withPolicyType("system_iceberg_orphan_file_removal")
+ .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()));
+ }
+
+ @Test
+ void createAndUpdateValidateRetentionUpperBoundary() throws Exception {
+ long maximum = IcebergOrphanFileRemovalContent.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\":\"cleanup\",\"policyType\":\"system_iceberg_orphan_file_removal\",\"content\":"
+ + content
+ + "}",
+ PolicyCreateRequest.class);
+ PolicyUpdateRequest update =
+ JsonUtils.objectMapper()
+ .readValue(
+
"{\"@type\":\"updateContent\",\"policyType\":\"system_iceberg_orphan_file_removal\",\"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/docs/open-api/policies.yaml b/docs/open-api/policies.yaml
index a079097ad5..2e0d953020 100644
--- a/docs/open-api/policies.yaml
+++ b/docs/open-api/policies.yaml
@@ -334,6 +334,111 @@ components:
items:
$ref: "#/components/schemas/TagForPolicyAssociation"
+ IcebergCompactionContent:
+ type: object
+ description: Built-in Iceberg compaction thresholds, scoring weights,
and job options.
+ properties:
+ minDataFileMse:
+ type: integer
+ format: int64
+ minimum: 0
+ default: 405323966463344
+ minDeleteFileNumber:
+ type: integer
+ format: int64
+ minimum: 0
+ default: 1
+ dataFileMseWeight:
+ type: integer
+ format: int64
+ minimum: 0
+ default: 1
+ deleteFileNumberWeight:
+ type: integer
+ format: int64
+ minimum: 0
+ default: 100
+ maxPartitionNum:
+ type: integer
+ format: int64
+ minimum: 1
+ default: 50
+ rewriteOptions:
+ type: object
+ default: {}
+ description: Rewrite options with nonempty values and keys matching
[A-Za-z0-9._-]+, without the job.options. prefix.
+ additionalProperties:
+ type: string
+ minLength: 1
+
+ IcebergCompactionPolicy:
+ allOf:
+ - $ref: '#/components/schemas/PolicyBase'
+ - type: object
+ required: [content]
+ properties:
+ policyType:
+ type: string
+ enum: [system_iceberg_compaction]
+ content:
+ $ref: '#/components/schemas/IcebergCompactionContent'
+
+ IcebergCompactionPolicyCreateRequest:
+ allOf:
+ - $ref: '#/components/schemas/PolicyCreateRequestBase'
+ - type: object
+ required: [content]
+ properties:
+ policyType:
+ type: string
+ enum: [system_iceberg_compaction]
+ content:
+ $ref: '#/components/schemas/IcebergCompactionContent'
+
+ IcebergOrphanFileRemovalContent:
+ type: object
+ description: Table-level orphan cleanup options. Scheduling is
configured separately.
+ properties:
+ olderThanDays:
+ type: integer
+ format: int64
+ minimum: 1
+ maximum: 36500
+ default: 3
+ description: Minimum age of orphan files in days, including for dry
runs.
+ location:
+ type: string
+ nullable: true
+ description: Optional scan location within the target table's
storage root.
+ dryRun:
+ type: boolean
+ default: false
+ description: List candidates without deleting them.
+
+ IcebergOrphanFileRemovalPolicy:
+ allOf:
+ - $ref: '#/components/schemas/PolicyBase'
+ - type: object
+ required: [content]
+ properties:
+ policyType:
+ type: string
+ enum: [system_iceberg_orphan_file_removal]
+ content:
+ $ref: '#/components/schemas/IcebergOrphanFileRemovalContent'
+
+ IcebergOrphanFileRemovalPolicyCreateRequest:
+ allOf:
+ - $ref: '#/components/schemas/PolicyCreateRequestBase'
+ - type: object
+ required: [content]
+ properties:
+ policyType:
+ type: string
+ enum: [system_iceberg_orphan_file_removal]
+ content:
+ $ref: '#/components/schemas/IcebergOrphanFileRemovalContent'
+
PolicyContentBase:
type: object
description: A base content structure for policies
@@ -417,10 +522,14 @@ components:
description: A basic policy without content
oneOf:
- $ref: '#/components/schemas/CustomPolicy'
+ - $ref: '#/components/schemas/IcebergOrphanFileRemovalPolicy'
+ - $ref: '#/components/schemas/IcebergCompactionPolicy'
discriminator:
propertyName: policyType
mapping:
custom: '#/components/schemas/CustomPolicy'
+ system_iceberg_orphan_file_removal:
'#/components/schemas/IcebergOrphanFileRemovalPolicy'
+ system_iceberg_compaction:
'#/components/schemas/IcebergCompactionPolicy'
PolicyListResponse:
type: object
@@ -496,10 +605,14 @@ components:
type: object
oneOf:
- $ref: '#/components/schemas/CustomPolicyCreateRequest'
+ - $ref:
'#/components/schemas/IcebergOrphanFileRemovalPolicyCreateRequest'
+ - $ref: '#/components/schemas/IcebergCompactionPolicyCreateRequest'
discriminator:
propertyName: policyType
mapping:
custom: '#/components/schemas/CustomPolicyCreateRequest'
+ system_iceberg_orphan_file_removal:
'#/components/schemas/IcebergOrphanFileRemovalPolicyCreateRequest'
+ system_iceberg_compaction:
'#/components/schemas/IcebergCompactionPolicyCreateRequest'
PolicyUpdatesRequest:
@@ -567,7 +680,7 @@ components:
"newComment": "This is my new policy comment"
}
- UpdatePolicyContentRequest:
+ UpdatePolicyContentRequestBase:
type: object
required:
- "@type"
@@ -576,20 +689,57 @@ components:
properties:
"@type":
type: string
- enum:
- - updateContent
+ enum: [updateContent]
policyType:
type: string
- description: The type of the policy
- newContent:
- type: object
- description: The new content of the policy, its structure depends on
the 'policyType'.
- oneOf:
- - $ref: '#/components/schemas/CustomPolicyContent'
- discriminator:
- propertyName: policyType
- mapping:
- custom: '#/components/schemas/CustomPolicyContent'
+ description: The type of the policy.
+
+ UpdateCustomPolicyContentRequest:
+ allOf:
+ - $ref: '#/components/schemas/UpdatePolicyContentRequestBase'
+ - type: object
+ properties:
+ policyType:
+ type: string
+ not:
+ enum: [system_iceberg_compaction,
system_iceberg_orphan_file_removal]
+ newContent:
+ $ref: '#/components/schemas/CustomPolicyContent'
+
+ UpdateIcebergCompactionPolicyContentRequest:
+ allOf:
+ - $ref: '#/components/schemas/UpdatePolicyContentRequestBase'
+ - type: object
+ properties:
+ policyType:
+ type: string
+ enum: [system_iceberg_compaction]
+ newContent:
+ $ref: '#/components/schemas/IcebergCompactionContent'
+
+ UpdateIcebergOrphanFileRemovalPolicyContentRequest:
+ allOf:
+ - $ref: '#/components/schemas/UpdatePolicyContentRequestBase'
+ - type: object
+ properties:
+ policyType:
+ type: string
+ enum: [system_iceberg_orphan_file_removal]
+ newContent:
+ $ref: '#/components/schemas/IcebergOrphanFileRemovalContent'
+
+ UpdatePolicyContentRequest:
+ description: Selects typed content using the sibling policyType field,
matching REST deserialization.
+ oneOf:
+ - $ref: '#/components/schemas/UpdateCustomPolicyContentRequest'
+ - $ref:
'#/components/schemas/UpdateIcebergCompactionPolicyContentRequest'
+ - $ref:
'#/components/schemas/UpdateIcebergOrphanFileRemovalPolicyContentRequest'
+ discriminator:
+ propertyName: policyType
+ mapping:
+ custom: '#/components/schemas/UpdateCustomPolicyContentRequest'
+ system_iceberg_compaction:
'#/components/schemas/UpdateIcebergCompactionPolicyContentRequest'
+ system_iceberg_orphan_file_removal:
'#/components/schemas/UpdateIcebergOrphanFileRemovalPolicyContentRequest'
example: {
"@type": "updateContent",
"policyType": "custom",
diff --git a/docs/table-maintenance-service/optimizer-configuration.md
b/docs/table-maintenance-service/optimizer-configuration.md
index f17eafc9ab..fe10b1321b 100644
--- a/docs/table-maintenance-service/optimizer-configuration.md
+++ b/docs/table-maintenance-service/optimizer-configuration.md
@@ -173,7 +173,7 @@ settings described above. Its job-specific `jobConf` keys
are:
| ------------------ |
-------------------------------------------------------------------------------------
| -------------------------------- |
| `catalog_name` | Iceberg catalog registered in Spark
| Required |
| `table_identifier` | Table identifier within the catalog, such as
`db.sample` | Required |
-| `older_than` | Timestamp in the Spark session time zone; must be at
least 24 hours old | Three days ago (Iceberg default) |
+| `older_than` | Timestamp (explicit offset or Spark session time zone);
must be at least 24 hours old | Three days ago (Iceberg default) |
| `location` | Scan only this directory within the table's storage
location | Table location |
| `dry_run` | `true` logs candidate paths without deleting; `false`
deletes | `false` |
| `spark_conf` | JSON string containing custom Spark settings, including
the Iceberg runtime if needed | None |
@@ -188,5 +188,63 @@ not supported. For a secured Iceberg REST catalog, supply
its authentication
settings explicitly in `spark_conf`, as described above.
See [Remove Orphan Files](./optimizer-cli-reference.md#remove-orphan-files)
for a
-complete submission example. Orphan cleanup has no built-in scheduling policy
-in this release.
+complete submission example. For policy-driven submission, configure the
handler
+and policy below. Periodic scheduling remains separate.
+
+### Orphan cleanup policy integration
+
+Register the handler alongside the existing optimizer providers:
+
+```properties
+gravitino.optimizer.strategyHandler.iceberg-orphan-file-removal.className =
org.apache.gravitino.maintenance.optimizer.recommender.handler.orphan.OrphanFileRemovalStrategyHandler
+```
+
+The built-in job adapter is registered automatically. Keep the same Spark and
+catalog submission configuration as for direct orphan cleanup, including
+`catalog_name` (the catalog alias configured in Spark) and `spark_conf`.
+The adapter supplies `table_identifier`, `older_than`, `location`, and
`dry_run`.
+These policy-derived values override shared submission defaults.
+
+Create a policy through `POST /api/metalakes/{metalake}/policies`:
+
+```json
+{
+ "name": "orphan_cleanup",
+ "policyType": "system_iceberg_orphan_file_removal",
+ "enabled": true,
+ "content": {
+ "olderThanDays": 3,
+ "dryRun": true
+ }
+}
+```
+
+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 3 and must be between 1 and 36500 inclusive
(approximately 100 years). This bound rejects impractical retention values
before a policy is stored. `dryRun` defaults to false.
+
+The policy adapter subtracts `olderThanDays` from the current instant and
always
+emits `older_than` as an explicit UTC timestamp ending in `Z`. Policy retention
+is therefore independent of `spark.sql.session.timeZone`. For direct job
+submission, timestamps without an offset use the Spark session time zone.
+
+Optional `location` must not have leading or trailing whitespace and must be
the
+table's storage root or a descendant; submission requires the table's
`location` metadata to validate a custom path. The Spark job
+rechecks containment and filesystem symlinks with its own credentials before
+listing or deleting files.
+
+The strategy operates on the whole table, including partitioned tables, and
does
+not need table or partition statistics. Each explicit optimizer invocation
makes
+an enabled, selected cleanup policy eligible with score 1. This does not
install
+a periodic scheduler, cooldown, or last-run tracking. Because the trigger is
always
+true and the score is always 1, every explicit `submit-strategy-jobs`
invocation
+for an enabled, selected orphan policy submits a Spark cleanup job. Unlike
+threshold-based compaction, there is no metric gate to suppress repeated runs.
+Operators must control the invocation rate through external scheduling. Start
+with `dryRun: true` to inspect candidates; dry-run jobs still consume Spark
+resources and also need rate limiting.
+
+Use `submit-strategy-jobs --strategy-name orphan_cleanup` with the target
+identifiers. The CLI `--dry-run` previews recommendations without submitting
jobs;
+the policy's `dryRun: true` submits a Spark job that lists candidates without
+deleting them. Start with that policy setting to inspect candidates.
diff --git a/docs/table-maintenance-service/optimizer.md
b/docs/table-maintenance-service/optimizer.md
index affd3f0b5b..5048ccb693 100644
--- a/docs/table-maintenance-service/optimizer.md
+++ b/docs/table-maintenance-service/optimizer.md
@@ -13,7 +13,7 @@ license: "This software is licensed under the Apache License
version 2."
The table maintenance service keeps tables healthy without anyone watching
them. Associate a policy with a tag and assign that tag to a table or one of
its ancestors. The service collects statistics, evaluates them against the
table's derived policy, and submits a job when the policy says work is needed.
-The framework is generic. Metrics collection, policy evaluation, and job
submission are not tied to any particular table format, and each is a Java
ServiceLoader extension point. What ships built in is deliberately narrower,
and in alpha that means Iceberg data file compaction on identity-partitioned
tables.
+The framework is generic. Metrics collection, policy evaluation, and job
submission are not tied to any particular table format, and each is a Java
ServiceLoader extension point. What ships built in is deliberately narrower,
and in alpha that includes Iceberg data file compaction on identity-partitioned
tables and table-level orphan cleanup.
The CLI binary, its configuration file, and its configuration keys carry the
older name `optimizer`, so you will see `gravitino-optimizer.sh`,
`gravitino-optimizer.conf`, and `gravitino.optimizer.*` throughout. Those are
literal strings rather than a second product.
@@ -21,7 +21,7 @@ The CLI binary, its configuration file, and its configuration
keys carry the old
Confirm your environment matches this list before starting an evaluation
against the built-ins. Anything outside it needs a custom extension, which is
covered in the [Extension Guide](./optimizer-extension-guide.md).
-- Compaction is the only built-in strategy. Snapshot expiration and orphan
file cleanup are available as directly submitted built-in jobs, but do not yet
have built-in scheduling strategies. Sort and cluster maintenance also require
custom strategies.
+- Compaction and orphan file cleanup have built-in strategies. Snapshot
expiration is available as a directly submitted built-in job. Scheduling and
minimum run intervals remain separate from strategy evaluation. Sort and
cluster maintenance also require custom strategies.
- Compaction applies to Iceberg tables only, and only where every partition
uses an identity transform.
- The service is driven through the CLI workflow rather than running on a
schedule of its own.
@@ -57,7 +57,12 @@ Three identifiers look interchangeable and are not.
`--strategy-name` takes the **policy name**, despite what it is called.
Passing either of the other two reports no matching identifiers rather than
naming the mistake.
-## Direct Orphan File Cleanup
+## Orphan File Cleanup
+
+For policy-driven submission, use `system_iceberg_orphan_file_removal` and
register
+the handler described in [Orphan cleanup policy
integration](./optimizer-configuration.md#orphan-cleanup-policy-integration).
+Each explicit optimizer invocation evaluates the enabled policy at table level;
+it does not require partition statistics or configure a periodic schedule.
Submit `builtin-iceberg-remove-orphan-files` through the jobs REST API to
reclaim
unreferenced files. Start with `dry_run: "true"` and review the candidate paths
@@ -65,7 +70,7 @@ in the job logs before allowing deletion. The default cutoff
is three days ago;
explicit cutoffs must be at least 24 hours old, including for dry runs. A
custom
scan location must remain within the target table's storage location.
-This is a directly submitted job, not a new scheduling strategy. See
+Direct submission remains available alongside policy-driven submission. See
[Remove Orphan Files](./optimizer-cli-reference.md#remove-orphan-files) for the
submission example and
[Configuration](./optimizer-configuration.md#orphan-file-cleanup-job-configuration)
for the per-job options.
diff --git
a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRemoveOrphanFilesJob.java
b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRemoveOrphanFilesJob.java
index 0aaab168a0..85c84075e8 100644
---
a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRemoveOrphanFilesJob.java
+++
b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRemoveOrphanFilesJob.java
@@ -18,6 +18,7 @@
*/
package org.apache.gravitino.maintenance.jobs.iceberg;
+import static
org.apache.gravitino.maintenance.optimizer.common.util.OrphanFileLocationUtils.normalizeLocation;
import static org.apache.spark.sql.functions.lit;
import com.google.common.base.Preconditions;
@@ -31,13 +32,13 @@ import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
-import java.util.Objects;
import java.util.stream.Stream;
import javax.annotation.Nullable;
import org.apache.gravitino.job.JobTemplateProvider;
import org.apache.gravitino.job.SparkJobTemplate;
import org.apache.gravitino.maintenance.jobs.BuiltInJob;
import
org.apache.gravitino.maintenance.optimizer.common.util.IcebergSparkConfigUtils;
+import
org.apache.gravitino.maintenance.optimizer.common.util.OrphanFileLocationUtils;
import org.apache.iceberg.Table;
import org.apache.iceberg.spark.Spark3Util;
import org.apache.spark.sql.AnalysisException;
@@ -100,7 +101,11 @@ public class IcebergRemoveOrphanFilesJob implements
BuiltInJob {
Table table =
Spark3Util.loadIcebergTable(
spark, IcebergJobUtils.escapeSqlIdentifier(catalog) + "." +
identifier);
- String location = options.getOrDefault("location", table.location());
+ String location = options.get("location");
+ // Direct callers may pass an empty template value without CLI argument
parsing.
+ if (location == null ||
location.codePoints().allMatch(Character::isWhitespace)) {
+ location = table.location();
+ }
validateLocation(table.location(), location);
validateLocalLocation(table.location(), location);
validateRemoteLocation(spark, table.location(), location);
@@ -158,19 +163,7 @@ public class IcebergRemoveOrphanFilesJob implements
BuiltInJob {
}
static void validateLocation(String tableLocation, String location) {
- URI root = normalizeLocation(tableLocation);
- URI requested = normalizeLocation(location);
- String rootPath = root.getPath().replaceAll("/+$", "");
- String childPath = requested.getPath().replaceAll("/+$", "");
- boolean sameStorage =
- Objects.equals(root.getScheme(), requested.getScheme())
- && Objects.equals(root.getAuthority(), requested.getAuthority());
- Preconditions.checkArgument(
- sameStorage
- && (childPath.equals(rootPath)
- || childPath.startsWith(rootPath.endsWith("/") ? rootPath :
rootPath + "/")),
- "location must be within the table's storage location: %s",
- tableLocation);
+ OrphanFileLocationUtils.validateLocation(tableLocation, location);
}
private static List<String> buildArguments() {
@@ -200,27 +193,6 @@ public class IcebergRemoveOrphanFilesJob implements
BuiltInJob {
+ "[--dry-run true|false] [--spark-conf <json>]");
}
- private static URI normalizeLocation(String value) {
- // Reject ambiguous encoded paths rather than allowing different
filesystem decoders to
- // interpret the containment check and the subsequent listing differently.
- Preconditions.checkArgument(
- !value.isEmpty() && !value.contains("%") && !value.contains("\\"),
- "Invalid scan location: %s",
- value);
- URI uri = URI.create(value);
- Preconditions.checkArgument(
- uri.getQuery() == null
- && uri.getFragment() == null
- && uri.getPath() != null
- && uri.getPath().startsWith("/"),
- "Scan location must be an absolute path without query or fragment: %s",
- value);
- if (uri.getScheme() == null || "file".equals(uri.getScheme())) {
- return (uri.getScheme() == null ? Paths.get(value) :
Paths.get(uri)).normalize().toUri();
- }
- return uri.normalize();
- }
-
private static void validateLocalLocation(String tableLocation, String
location)
throws IOException {
URI requested = normalizeLocation(location);
diff --git
a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRemoveOrphanFilesJobWithSpark.java
b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRemoveOrphanFilesJobWithSpark.java
index 1d3a8d821e..f1e6c59626 100644
---
a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRemoveOrphanFilesJobWithSpark.java
+++
b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRemoveOrphanFilesJobWithSpark.java
@@ -96,6 +96,26 @@ public class TestIcebergRemoveOrphanFilesJobWithSpark {
assertEquals(0, IcebergRemoveOrphanFilesJob.execute(spark, args));
}
+ /** Verifies blank locations default to the table root without CLI argument
parsing. */
+ @Test
+ public void testBlankLocationDefaultsToTableRoot() throws Exception {
+ spark.sql("CREATE TABLE test_catalog.db.blank_location (id INT) USING
iceberg");
+ Path root = tempDir.resolve("db/blank_location");
+ Path old = Files.write(root.resolve("old-orphan"), new byte[] {1});
+ Files.setLastModifiedTime(old, FileTime.from(Instant.now().minus(5,
ChronoUnit.DAYS)));
+ Map<String, String> options = args("blank_location");
+ options.put("dry-run", "true");
+ for (String location : new String[] {"", " ", "\t", "\u2003"}) {
+ options.put("location", location);
+ assertEquals(1, IcebergRemoveOrphanFilesJob.execute(spark, options));
+ assertTrue(Files.exists(old));
+ }
+ options.put("location", "");
+ options.put("dry-run", "false");
+ assertEquals(1, IcebergRemoveOrphanFilesJob.execute(spark, options));
+ assertFalse(Files.exists(old));
+ }
+
/** Verifies custom location and outside rejection. */
@Test
public void testCustomLocationAndOutsideRejection() throws Exception {
@@ -123,6 +143,29 @@ public class TestIcebergRemoveOrphanFilesJobWithSpark {
IllegalArgumentException.class, () ->
IcebergRemoveOrphanFilesJob.execute(spark, args));
}
+ /** Verifies UTC cutoffs from the policy adapter retain their meaning in
other session zones. */
+ @Test
+ public void testUtcCutoffWithNonUtcSession() throws Exception {
+ spark.sql("CREATE TABLE test_catalog.db.utc_cutoff (id INT) USING
iceberg");
+ Path root = tempDir.resolve("db/utc_cutoff");
+ Instant cutoff = Instant.now().minus(4,
ChronoUnit.DAYS).truncatedTo(ChronoUnit.SECONDS);
+ Path older = Files.write(root.resolve("older"), new byte[] {1});
+ Path newer = Files.write(root.resolve("newer"), new byte[] {1});
+ Files.setLastModifiedTime(older, FileTime.from(cutoff.minusSeconds(60)));
+ Files.setLastModifiedTime(newer, FileTime.from(cutoff.plusSeconds(60)));
+ Map<String, String> options = args("utc_cutoff");
+ options.put("older-than", cutoff.toString().replace("T", " "));
+ String timezone = spark.conf().get("spark.sql.session.timeZone");
+ spark.conf().set("spark.sql.session.timeZone", "America/Los_Angeles");
+ try {
+ assertEquals(1, IcebergRemoveOrphanFilesJob.execute(spark, options));
+ assertFalse(Files.exists(older));
+ assertTrue(Files.exists(newer));
+ } finally {
+ spark.conf().set("spark.sql.session.timeZone", timezone);
+ }
+ }
+
/** Verifies invalid inputs fail before deletion. */
@Test
public void testInvalidInputsFailBeforeDeletion() throws Exception {
diff --git
a/maintenance/optimizer-api/src/main/java/org/apache/gravitino/maintenance/optimizer/common/util/OrphanFileLocationUtils.java
b/maintenance/optimizer-api/src/main/java/org/apache/gravitino/maintenance/optimizer/common/util/OrphanFileLocationUtils.java
new file mode 100644
index 0000000000..c09ea3bc7c
--- /dev/null
+++
b/maintenance/optimizer-api/src/main/java/org/apache/gravitino/maintenance/optimizer/common/util/OrphanFileLocationUtils.java
@@ -0,0 +1,79 @@
+/*
+ * 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.common.util;
+
+import com.google.common.base.Preconditions;
+import java.net.URI;
+import java.nio.file.Paths;
+import java.util.Objects;
+
+/** Lexical containment checks shared by orphan cleanup submission and
execution. */
+public final class OrphanFileLocationUtils {
+ private OrphanFileLocationUtils() {}
+
+ /**
+ * Validates that a scan is within the table root. Filesystem symlink
validation must also run in
+ * the job, using its storage credentials and filesystem view.
+ *
+ * @param tableLocation table storage root
+ * @param location requested scan location
+ */
+ public static void validateLocation(String tableLocation, String location) {
+ URI root = normalizeLocation(tableLocation);
+ URI requested = normalizeLocation(location);
+ String rootPath = root.getPath().replaceAll("/+$", "");
+ String childPath = requested.getPath().replaceAll("/+$", "");
+ boolean sameStorage =
+ Objects.equals(root.getScheme(), requested.getScheme())
+ && Objects.equals(root.getAuthority(), requested.getAuthority());
+ Preconditions.checkArgument(
+ sameStorage
+ && (childPath.equals(rootPath)
+ || childPath.startsWith(rootPath.endsWith("/") ? rootPath :
rootPath + "/")),
+ "location must be within the table's storage location: %s",
+ tableLocation);
+ }
+
+ /**
+ * Normalizes an absolute location and rejects ambiguous URI encodings.
+ *
+ * @param value input location
+ * @return normalized URI
+ */
+ public static URI normalizeLocation(String value) {
+ // Reject ambiguous encoded paths rather than allowing different
filesystem decoders to
+ // interpret the containment check and the subsequent listing differently.
+ Preconditions.checkArgument(
+ !value.isEmpty() && !value.contains("%") && !value.contains("\\"),
+ "Invalid scan location: %s",
+ value);
+ URI uri = URI.create(value);
+ Preconditions.checkArgument(
+ uri.getQuery() == null
+ && uri.getFragment() == null
+ && uri.getPath() != null
+ && uri.getPath().startsWith("/"),
+ "Scan location must be an absolute path without query or fragment: %s",
+ value);
+ if (uri.getScheme() == null || "file".equals(uri.getScheme())) {
+ return (uri.getScheme() == null ? Paths.get(value) :
Paths.get(uri)).normalize().toUri();
+ }
+ return uri.normalize();
+ }
+}
diff --git a/maintenance/optimizer/build.gradle.kts
b/maintenance/optimizer/build.gradle.kts
index abb18c589b..aa6b482204 100644
--- a/maintenance/optimizer/build.gradle.kts
+++ b/maintenance/optimizer/build.gradle.kts
@@ -132,6 +132,9 @@ tasks.test {
} else {
dependsOn(tasks.jar)
dependsOn(":server:jar")
+ // Embedded optimizer integration tests load the Iceberg catalog in an
isolated classloader.
+ dependsOn(":catalogs:catalog-lakehouse-iceberg:jar")
+ dependsOn(":catalogs:catalog-lakehouse-iceberg:runtimeJars")
}
}
diff --git
a/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/BaseExpressionStrategyHandler.java
b/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/BaseExpressionStrategyHandler.java
index 07072adfda..38230144aa 100644
---
a/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/BaseExpressionStrategyHandler.java
+++
b/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/BaseExpressionStrategyHandler.java
@@ -127,7 +127,10 @@ public abstract class BaseExpressionStrategyHandler
implements StrategyHandler {
}
private boolean isPartitionTable() {
- return tableMetadata.partitioning().length > 0;
+ // Table-level handlers must also work on partitioned tables without
requesting partition
+ // statistics. Only handlers declaring PARTITION_STATISTICS use
per-partition evaluation.
+ return dataRequirements().contains(DataRequirement.PARTITION_STATISTICS)
+ && tableMetadata.partitioning().length > 0;
}
private boolean shouldTriggerForPartitionTable() {
diff --git
a/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/orphan/OrphanFileRemovalJobContext.java
b/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/orphan/OrphanFileRemovalJobContext.java
new file mode 100644
index 0000000000..4999f8bc9b
--- /dev/null
+++
b/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/orphan/OrphanFileRemovalJobContext.java
@@ -0,0 +1,74 @@
+/*
+ * 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.orphan;
+
+import java.util.Map;
+import javax.annotation.Nullable;
+import org.apache.gravitino.NameIdentifier;
+import
org.apache.gravitino.maintenance.optimizer.api.recommender.JobExecutionContext;
+
+/** Immutable target and options for table-level orphan cleanup. */
+public class OrphanFileRemovalJobContext implements JobExecutionContext {
+ private final NameIdentifier name;
+ private final Map<String, String> jobOptions;
+ private final String jobTemplateName;
+ @Nullable private final String tableLocation;
+
+ /**
+ * 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
+ * @param tableLocation table storage root, if available
+ */
+ public OrphanFileRemovalJobContext(
+ NameIdentifier name,
+ Map<String, String> jobOptions,
+ String jobTemplateName,
+ @Nullable String tableLocation) {
+ this.name = name;
+ this.jobOptions = Map.copyOf(jobOptions);
+ this.jobTemplateName = jobTemplateName;
+ this.tableLocation = tableLocation;
+ }
+
+ @Override
+ public NameIdentifier nameIdentifier() {
+ return name;
+ }
+
+ @Override
+ public Map<String, String> jobOptions() {
+ return jobOptions;
+ }
+
+ @Override
+ public String jobTemplateName() {
+ return jobTemplateName;
+ }
+
+ /**
+ * @return table storage root, or null if absent from metadata
+ */
+ @Nullable
+ public String tableLocation() {
+ return tableLocation;
+ }
+}
diff --git
a/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/orphan/OrphanFileRemovalStrategyHandler.java
b/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/orphan/OrphanFileRemovalStrategyHandler.java
new file mode 100644
index 0000000000..d115957a9a
--- /dev/null
+++
b/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/orphan/OrphanFileRemovalStrategyHandler.java
@@ -0,0 +1,65 @@
+/*
+ * 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.orphan;
+
+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.IcebergOrphanFileRemovalContent;
+import org.apache.gravitino.rel.Table;
+
+/** Evaluates orphan cleanup at table level, including for partitioned tables.
*/
+public class OrphanFileRemovalStrategyHandler extends
BaseExpressionStrategyHandler {
+ // Iceberg exposes its storage root through the catalog's "location" table
property.
+ private static final String TABLE_LOCATION_PROPERTY = "location";
+
+ /** Registered strategy type. */
+ public static final String NAME =
IcebergOrphanFileRemovalContent.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 OrphanFileRemovalJobContext(
+ nameIdentifier,
+ jobOptions,
+ strategy.jobTemplateName(),
+ tableMetadata.properties() == null
+ ? null
+ : tableMetadata.properties().get(TABLE_LOCATION_PROPERTY));
+ }
+}
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 a6b9362c80..7b1cceeaf5 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
@@ -31,6 +31,7 @@ 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.GravitinoClientUtils;
import org.apache.gravitino.policy.IcebergDataCompactionContent;
+import org.apache.gravitino.policy.IcebergOrphanFileRemovalContent;
/** Submits optimizer jobs to Gravitino using job template adapters. */
public class GravitinoJobSubmitter implements JobSubmitter {
@@ -49,7 +50,9 @@ public class GravitinoJobSubmitter implements JobSubmitter {
private final Map<String, Class<? extends GravitinoJobAdapter>> jobAdapters =
ImmutableMap.of(
IcebergDataCompactionContent.JOB_TEMPLATE_NAME_VALUE,
- GravitinoCompactionJobAdapter.class);
+ GravitinoCompactionJobAdapter.class,
+ IcebergOrphanFileRemovalContent.JOB_TEMPLATE_NAME_VALUE,
+ GravitinoOrphanFileRemovalJobAdapter.class);
@Override
public String name() {
diff --git
a/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/job/GravitinoOrphanFileRemovalJobAdapter.java
b/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/job/GravitinoOrphanFileRemovalJobAdapter.java
new file mode 100644
index 0000000000..8ce94dfaea
--- /dev/null
+++
b/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/job/GravitinoOrphanFileRemovalJobAdapter.java
@@ -0,0 +1,104 @@
+/*
+ * 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.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.common.util.OrphanFileLocationUtils;
+import
org.apache.gravitino.maintenance.optimizer.recommender.handler.orphan.OrphanFileRemovalJobContext;
+import org.apache.gravitino.policy.IcebergOrphanFileRemovalContent;
+
+/** Converts orphan cleanup options to the existing Spark template's
configuration. */
+public class GravitinoOrphanFileRemovalJobAdapter implements
GravitinoJobAdapter {
+ private static final String TABLE_IDENTIFIER_KEY = "table_identifier";
+ private static final String OLDER_THAN_KEY = "older_than";
+ private static final String LOCATION_KEY = "location";
+ private static final String DRY_RUN_KEY = "dry_run";
+ 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 GravitinoOrphanFileRemovalJobAdapter() {
+ this(Clock.systemUTC());
+ }
+
+ GravitinoOrphanFileRemovalJobAdapter(Clock clock) {
+ this.clock = clock;
+ }
+
+ @Override
+ public Map<String, String> jobConfig(JobExecutionContext context) {
+ Preconditions.checkArgument(
+ context instanceof OrphanFileRemovalJobContext,
+ "jobExecutionContext must be OrphanFileRemovalJobContext");
+ OrphanFileRemovalJobContext orphan = (OrphanFileRemovalJobContext) context;
+ Map<String, String> options = orphan.jobOptions();
+ long days;
+ try {
+ days =
+ Long.parseLong(
+ options.getOrDefault(
+ IcebergOrphanFileRemovalContent.OLDER_THAN_DAYS_KEY,
+
String.valueOf(IcebergOrphanFileRemovalContent.DEFAULT_OLDER_THAN_DAYS)));
+ } catch (NumberFormatException e) {
+ throw new IllegalArgumentException("olderThanDays must be an integer",
e);
+ }
+ IcebergOrphanFileRemovalContent.validateOlderThanDays(days);
+ String dryRun =
+ options.getOrDefault(
+ IcebergOrphanFileRemovalContent.DRY_RUN_KEY,
+ String.valueOf(IcebergOrphanFileRemovalContent.DEFAULT_DRY_RUN));
+ Preconditions.checkArgument(
+ "true".equals(dryRun) || "false".equals(dryRun), "dryRun must be true
or false");
+ String location =
options.getOrDefault(IcebergOrphanFileRemovalContent.LOCATION_KEY, "");
+ if (options.containsKey(IcebergOrphanFileRemovalContent.LOCATION_KEY)) {
+ Preconditions.checkArgument(
+ orphan.tableLocation() != null,
+ "Table location is required to validate a custom scan location");
+ OrphanFileLocationUtils.validateLocation(orphan.tableLocation(),
location);
+ location =
OrphanFileLocationUtils.normalizeLocation(location).toString();
+ }
+ return Map.of(
+ TABLE_IDENTIFIER_KEY,
+
IdentifierUtils.removeCatalogFromIdentifier(orphan.nameIdentifier()).toString(),
+ OLDER_THAN_KEY,
+ cutoff(days),
+ LOCATION_KEY,
+ location,
+ DRY_RUN_KEY,
+ dryRun);
+ }
+
+ 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 a cleanup timestamp", e);
+ }
+ }
+}
diff --git
a/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/integration/test/RecommenderIT.java
b/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/integration/test/RecommenderIT.java
index 9076719ad0..ea72f69c7b 100644
---
a/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/integration/test/RecommenderIT.java
+++
b/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/integration/test/RecommenderIT.java
@@ -20,7 +20,14 @@
package org.apache.gravitino.maintenance.optimizer.integration.test;
import com.google.common.collect.ImmutableMap;
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
import java.util.Arrays;
+import java.util.Base64;
import java.util.List;
import java.util.Map;
import java.util.UUID;
@@ -37,8 +44,12 @@ import
org.apache.gravitino.maintenance.optimizer.common.conf.OptimizerConfig;
import org.apache.gravitino.maintenance.optimizer.recommender.Recommender;
import
org.apache.gravitino.maintenance.optimizer.recommender.handler.compaction.CompactionJobContext;
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.job.GravitinoOrphanFileRemovalJobAdapter;
import
org.apache.gravitino.maintenance.optimizer.recommender.util.StrategyUtils;
import
org.apache.gravitino.maintenance.optimizer.updater.statistics.GravitinoStatisticsUpdater;
+import org.apache.gravitino.policy.IcebergOrphanFileRemovalContent;
+import org.apache.gravitino.policy.PolicyContents;
import org.apache.gravitino.stats.StatisticValues;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.Assertions;
@@ -67,6 +78,11 @@ public class RecommenderIT extends
AbstractGravitinoOptimizerEnvIT {
+ CompactionStrategyHandler.NAME
+ ".className",
CompactionStrategyHandler.class.getName(),
+ OptimizerConfig.OPTIMIZER_PREFIX
+ + "strategyHandler."
+ + OrphanFileRemovalStrategyHandler.NAME
+ + ".className",
+ OrphanFileRemovalStrategyHandler.class.getName(),
OptimizerConfig.JOB_SUBMITTER_CONFIG.getKey(),
RecordingJobSubmitterForIT.NAME,
RecordingJobSubmitterForIT.SESSION_ID_KEY,
@@ -86,6 +102,141 @@ public class RecommenderIT extends
AbstractGravitinoOptimizerEnvIT {
}
}
+ @Test
+ void testOrphanCleanupRejectsLocationWhitespace() throws Exception {
+ String name = "location_cleanup";
+ String type = "system_iceberg_orphan_file_removal";
+ String validLocation = "s3://bucket/table/data";
+ metalakeClient.createPolicy(
+ name,
+ type,
+ "valid location",
+ true,
+ PolicyContents.icebergOrphanFileRemoval(3, validLocation, true));
+ for (String location :
+ new String[] {
+ " " + validLocation,
+ validLocation + " ",
+ "\u2003" + validLocation,
+ validLocation + "\u2003"
+ }) {
+ HttpResponse<String> create =
+ policyRequest(
+ "POST",
+ "",
+ "{\"name\":\"invalid_location_cleanup\",\"policyType\":\""
+ + type
+ + "\",\"enabled\":true,\"content\":{\"location\":\""
+ + location
+ + "\"}}");
+ Assertions.assertEquals(400, create.statusCode(), create.body());
+ Assertions.assertTrue(create.body().contains("location"));
+ HttpResponse<String> update =
+ policyRequest(
+ "PUT",
+ "/" + name,
+ "{\"updates\":[{\"@type\":\"updateContent\",\"policyType\":\""
+ + type
+ + "\",\"newContent\":{\"location\":\""
+ + location
+ + "\"}}]}");
+ Assertions.assertEquals(400, update.statusCode(), update.body());
+ Assertions.assertTrue(update.body().contains("location"));
+ Assertions.assertEquals(
+ validLocation,
+
metalakeClient.getPolicy(name).content().rules().get("job.options.location"));
+ }
+ }
+
+ @Test
+ void testOrphanCleanupRejectsOverflowingRetention() throws Exception {
+ HttpResponse<String> response =
+ policyRequest(
+ "POST",
+ "",
+
"{\"name\":\"overflowing_cleanup\",\"policyType\":\"system_iceberg_orphan_file_removal\","
+ + "\"enabled\":true,\"content\":{\"olderThanDays\":"
+ + Long.MAX_VALUE
+ + "}}");
+ Assertions.assertEquals(400, response.statusCode(), response.body());
+ Assertions.assertTrue(response.body().contains("olderThanDays"));
+ }
+
+ @Test
+ void testOrphanCleanupRejectsInvalidUpdateAndPreservesPolicy() throws
Exception {
+ String name = "bounded_cleanup";
+ String type = "system_iceberg_orphan_file_removal";
+ long maximum = IcebergOrphanFileRemovalContent.MAX_OLDER_THAN_DAYS;
+ metalakeClient.createPolicy(
+ name,
+ type,
+ "maximum retention",
+ true,
+ PolicyContents.icebergOrphanFileRemoval(maximum, null, true));
+ HttpResponse<String> response =
+ policyRequest(
+ "PUT",
+ "/" + name,
+ "{\"updates\":[{\"@type\":\"updateContent\",\"policyType\":\""
+ + type
+ + "\",\"newContent\":{\"olderThanDays\":"
+ + (maximum + 1)
+ + "}}]}");
+ Assertions.assertEquals(400, response.statusCode(), response.body());
+ Assertions.assertTrue(response.body().contains("olderThanDays"));
+ Assertions.assertEquals(
+ String.valueOf(maximum),
+ metalakeClient
+ .getPolicy(name)
+ .content()
+ .rules()
+ .get(
+ IcebergOrphanFileRemovalContent.JOB_OPTIONS_PREFIX
+ + IcebergOrphanFileRemovalContent.OLDER_THAN_DAYS_KEY)
+ .toString());
+ }
+
+ @Test
+ void testOrphanCleanupPolicyThroughRestAndTags() throws Exception {
+ String tableName = "orphan_cleanup_partitioned";
+ String policyName = "orphan_cleanup_policy";
+ createPartitionTable(tableName);
+ metalakeClient.createPolicy(
+ policyName,
+ "system_iceberg_orphan_file_removal",
+ "cleanup",
+ true,
+ PolicyContents.icebergOrphanFileRemoval(3, null, true));
+ createTagForPolicy(policyName);
+ associatePolicyTagToTable(policyName, tableName);
+ try (Recommender recommender = new Recommender(optimizerEnv)) {
+ List<JobExecutionContext> jobs =
+ recommendForOneStrategy(recommender,
List.of(getTableIdentifier(tableName)), policyName);
+ Assertions.assertEquals(1, jobs.size());
+ Assertions.assertEquals("builtin-iceberg-remove-orphan-files",
jobs.get(0).jobTemplateName());
+ Map<String, String> config =
+ new GravitinoOrphanFileRemovalJobAdapter().jobConfig(jobs.get(0));
+ Assertions.assertEquals(TEST_SCHEMA + "." + tableName,
config.get("table_identifier"));
+ Assertions.assertEquals("true", config.get("dry_run"));
+ Assertions.assertEquals("", config.get("location"));
+ }
+ metalakeClient.disablePolicy(policyName);
+ RecordingJobSubmitterForIT.reset(SESSION_ID);
+ try (Recommender recommender = new Recommender(optimizerEnv)) {
+ IllegalArgumentException exception =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ recommender.submitForStrategyName(
+ List.of(getTableIdentifier(tableName)), policyName));
+ Assertions.assertTrue(
+ exception.getMessage().contains("No identifiers matched strategy
name"));
+
Assertions.assertTrue(RecordingJobSubmitterForIT.submittedContexts(SESSION_ID).isEmpty());
+ } finally {
+ RecordingJobSubmitterForIT.clear(SESSION_ID);
+ }
+ }
+
@Test
void testRecommendNonPartitionTable() throws Exception {
String policyForDelete = "policyForDelete";
@@ -271,6 +422,24 @@ public class RecommenderIT extends
AbstractGravitinoOptimizerEnvIT {
}
}
+ private HttpResponse<String> policyRequest(String method, String suffix,
String body)
+ throws Exception {
+ String credentials =
+ Base64.getEncoder()
+ .encodeToString(
+ (System.getProperty("user.name") +
":").getBytes(StandardCharsets.UTF_8));
+ HttpRequest request =
+ HttpRequest.newBuilder(
+ URI.create(serverUri + "/api/metalakes/" + METALAKE_NAME +
"/policies" + suffix))
+ .timeout(Duration.ofSeconds(30))
+ .header("Content-Type", "application/json")
+ .header("Accept", "application/vnd.gravitino.v1+json")
+ .header("Authorization", "Basic " + credentials)
+ .method(method, HttpRequest.BodyPublishers.ofString(body))
+ .build();
+ return HttpClient.newHttpClient().send(request,
HttpResponse.BodyHandlers.ofString());
+ }
+
private List<JobExecutionContext> recommendForOneStrategy(
Recommender recommender, List<NameIdentifier> identifiers, String
strategyName) {
RecordingJobSubmitterForIT.reset(SESSION_ID);
diff --git
a/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/recommender/job/TestGravitinoJobSubmitter.java
b/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/recommender/job/TestGravitinoJobSubmitter.java
index 68f79671c4..2963fd003c 100644
---
a/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/recommender/job/TestGravitinoJobSubmitter.java
+++
b/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/recommender/job/TestGravitinoJobSubmitter.java
@@ -25,6 +25,7 @@ import
org.apache.gravitino.maintenance.optimizer.api.recommender.JobExecutionCo
import org.apache.gravitino.maintenance.optimizer.common.OptimizerEnv;
import org.apache.gravitino.maintenance.optimizer.common.conf.OptimizerConfig;
import org.apache.gravitino.policy.IcebergDataCompactionContent;
+import org.apache.gravitino.policy.IcebergOrphanFileRemovalContent;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -37,6 +38,14 @@ public class TestGravitinoJobSubmitter {
Assertions.assertTrue(adapter instanceof GravitinoCompactionJobAdapter);
}
+ @Test
+ void loadJobAdapterReturnsOrphanAdapterForBuiltInTemplateName() {
+ GravitinoJobSubmitter submitter = new GravitinoJobSubmitter();
+ GravitinoJobAdapter adapter =
+
submitter.loadJobAdapter(IcebergOrphanFileRemovalContent.JOB_TEMPLATE_NAME_VALUE);
+ Assertions.assertTrue(adapter instanceof
GravitinoOrphanFileRemovalJobAdapter);
+ }
+
@Test
void loadJobAdapterFallsBackToConfiguredClassName() {
String jobTemplateName = "custom";
diff --git
a/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/recommender/job/TestGravitinoOrphanFileRemovalJobAdapter.java
b/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/recommender/job/TestGravitinoOrphanFileRemovalJobAdapter.java
new file mode 100644
index 0000000000..cce5600303
--- /dev/null
+++
b/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/recommender/job/TestGravitinoOrphanFileRemovalJobAdapter.java
@@ -0,0 +1,159 @@
+/*
+ * 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.orphan.OrphanFileRemovalJobContext;
+import
org.apache.gravitino.maintenance.optimizer.recommender.handler.orphan.OrphanFileRemovalStrategyHandler;
+import
org.apache.gravitino.maintenance.optimizer.recommender.strategy.GravitinoStrategy;
+import org.apache.gravitino.policy.IcebergOrphanFileRemovalContent;
+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 TestGravitinoOrphanFileRemovalJobAdapter {
+ private static final String TEMPLATE =
IcebergOrphanFileRemovalContent.JOB_TEMPLATE_NAME_VALUE;
+ private static final Clock CLOCK =
+ Clock.fixed(Instant.parse("2026-09-27T12:30:00Z"), ZoneOffset.UTC);
+ private final GravitinoOrphanFileRemovalJobAdapter adapter =
+ new GravitinoOrphanFileRemovalJobAdapter(CLOCK);
+
+ @Test
+ void policyToSubmissionWorksWithoutPartitionStatistics() {
+ for (boolean partitioned : new boolean[] {false, true}) {
+ Policy policy = Mockito.mock(Policy.class);
+ Mockito.when(policy.content())
+ .thenReturn(PolicyContents.icebergOrphanFileRemoval(3,
"s3://bucket/table/data", true));
+ GravitinoStrategy strategy = new GravitinoStrategy(policy);
+ Table table = Mockito.mock(Table.class);
+ Mockito.when(table.properties()).thenReturn(Map.of("location",
"s3://bucket/table"));
+ Mockito.when(table.partitioning())
+ .thenReturn(partitioned ? new Transform[]
{Transforms.identity("id")} : new Transform[0]);
+ OrphanFileRemovalStrategyHandler handler = new
OrphanFileRemovalStrategyHandler();
+ 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());
+ OrphanFileRemovalJobContext context =
+ (OrphanFileRemovalJobContext)
handler.evaluate().jobExecutionContext().orElseThrow();
+ Assertions.assertInstanceOf(
+ GravitinoOrphanFileRemovalJobAdapter.class,
+ new
GravitinoJobSubmitter().loadJobAdapter(context.jobTemplateName()));
+ OptimizerConfig config =
+ new OptimizerConfig(
+ Map.of(
+ OptimizerConfig.JOB_SUBMITTER_CONFIG_PREFIX +
"catalog_name", "spark_catalog",
+ OptimizerConfig.JOB_SUBMITTER_CONFIG_PREFIX + "dry_run",
"false"));
+ 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"));
+ Assertions.assertEquals("2026-09-24 12:30:00Z", job.get("older_than"));
+ Assertions.assertEquals("true", job.get("dry_run"));
+ Assertions.assertEquals("s3://bucket/table/data", job.get("location"));
+ }
+ }
+
+ @Test
+ void defaultsFillOptionalPlaceholders() {
+ Map<String, String> job = adapter.jobConfig(context(Map.of(), null));
+ Assertions.assertEquals("", job.get("location"));
+ Assertions.assertEquals("false", job.get("dry_run"));
+ Assertions.assertEquals("2026-09-24 12:30:00Z", job.get("older_than"));
+ Assertions.assertEquals(
+ "2026-09-26 12:30:00Z",
+ adapter.jobConfig(context(Map.of("olderThanDays", "1"),
null)).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),
null)));
+ }
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> adapter.jobConfig(context(Map.of("dryRun", "yes"), null)));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> adapter.jobConfig(context(Map.of("location",
"s3://bucket/table"), null)));
+ for (String location :
+ new String[] {
+ "s3://bucket/table-other",
+ "s3://bucket/table/../other",
+ "s3://other/table",
+ "s3://bucket/table/%2e%2e/other",
+ "s3://bucket/table?query",
+ ""
+ }) {
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> adapter.jobConfig(context(Map.of("location", location),
"s3://bucket/table")));
+ }
+ }
+
+ @Test
+ void validatesRetentionRangeAndReportsTimestampErrors() {
+ long maximum = IcebergOrphanFileRemovalContent.MAX_OLDER_THAN_DAYS;
+ Assertions.assertEquals(
+ "1926-10-22 12:30:00Z",
+ adapter
+ .jobConfig(context(Map.of("olderThanDays",
String.valueOf(maximum)), null))
+ .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),
null)));
+ Assertions.assertTrue(error.getMessage().contains("olderThanDays"));
+ }
+ GravitinoOrphanFileRemovalJobAdapter ancient =
+ new GravitinoOrphanFileRemovalJobAdapter(Clock.fixed(Instant.MIN,
ZoneOffset.UTC));
+ IllegalArgumentException error =
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () ->
ancient.jobConfig(context(Map.of(), null)));
+ Assertions.assertTrue(error.getMessage().contains("olderThanDays"));
+ Assertions.assertNotNull(error.getCause());
+ }
+
+ private OrphanFileRemovalJobContext context(Map<String, String> options,
String location) {
+ return new OrphanFileRemovalJobContext(
+ NameIdentifier.of("catalog", "db", "table"), options, TEMPLATE,
location);
+ }
+}
