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 38bc5dd156 [#11196] feat(maintenance): Trigger Iceberg manifest
rewrites from policies (#13533)
38bc5dd156 is described below
commit 38bc5dd1560e6dd592b787ae1e8d6f9cd663ffb9
Author: Akshay Thorat <[email protected]>
AuthorDate: Sun Oct 25 19:39:04 2026 -0700
[#11196] feat(maintenance): Trigger Iceberg manifest rewrites from policies
(#13533)
### What changes were proposed in this pull request?
Add configurable thresholds, a built-in strategy handler, and a job
adapter for `system_iceberg_rewrite_manifests`, using the collector
merged in #13717. Trigger at 500 manifests, or at 100 with average size
below 8 MiB.
Resolve omitted `spec_id` from Iceberg's current
`TableMetadata.defaultSpecId()`, projected as reserved, read-only
`default-spec-id` table metadata. Explicit IDs override the default. Pin
the resolved ID through job submission so later partition evolution
cannot change the rewrite target.
Require complete count/average statistics and compare their snapshot ID
with metadata fetched after statistics. Validate integer and boolean
policy fields without coercion. Invalid policy/statistics inputs or an
unresolvable default skip the affected table while the batch continues.
### Why are the changes needed?
Connect manifest statistics to policy-driven rewrites, support empty
policy content, and preserve the evaluated partition spec through
submission. Related to #11196 and design #12942.
### Does this PR introduce _any_ user-facing change?
Adds count/size thresholds, optional `spec_id`, and `use_caching` to the
typed policy. Users do not configure `default-spec-id`; the catalog
exposes it from Iceberg metadata. Custom metadata providers must expose
it for policies without an explicit spec.
Missing/incomplete statistics or unverifiable snapshot metadata produce
no recommendation. Recollect after snapshot changes. Policy options
override shared submitter settings; omitted caching preserves the
Iceberg runtime default. Snapshot checks are not transactional with
submission. Scheduling and cooldowns remain separate.
### How was this patch tested?
- Full Iceberg catalog and optimizer suites, plus focused API, DTO, and
rewrite-manifests job tests: 474 passed, 5 skipped in the latest run.
- Real Spark regression reproduces the omitted-spec failure, evaluates
spec 1, evolves the default to spec 2 before submission, verifies only
spec 1 is rewritten, and selects spec 2 on the next evaluation.
- Regressions cover explicit overrides, missing/malformed defaults,
healthy tables continuing in the batch, metadata projection after
evolution, and rejection of writes to the reserved property.
- Policy REST validation, caching overrides, snapshot freshness, and
invalid handler inputs were also tested in earlier revisions.
- `spotlessApply`, affected-module Javadocs, `:docs:build`, `rat`, and
`git diff --check` passed.
Docker-tagged tests were not run locally because the macOS Docker
connector is unavailable.
---------
Co-authored-by: İbrahim ERBİLEN <[email protected]>
---
.../policy/IcebergRewriteManifestsContent.java | 158 +++++++--
.../apache/gravitino/policy/PolicyContents.java | 32 +-
.../policy/TestIcebergRewriteManifestsContent.java | 66 ++++
.../lakehouse/iceberg/IcebergConstants.java | 4 +
.../iceberg/IcebergTablePropertiesMetadata.java | 6 +
.../iceberg/ops/IcebergCatalogWrapperHelper.java | 3 +
.../iceberg/utils/IcebergTablePropertiesUtil.java | 2 +
.../TestIcebergTablePropertiesMetadata.java | 15 +
.../iceberg/ops/TestIcebergTableUpdate.java | 11 +
.../utils/TestIcebergTablePropertiesUtil.java | 56 +++
.../gravitino/dto/policy/PolicyContentDTO.java | 133 ++++++-
.../apache/gravitino/dto/util/DTOConverters.java | 18 +-
.../dto/policy/TestManifestRewritePolicyDTO.java | 154 +++++++++
docs/iceberg-rewrite-manifests-policy.md | 105 ++++++
docs/lakehouse-iceberg-catalog.md | 1 +
docs/open-api/policies.yaml | 31 +-
docs/policies.md | 20 +-
.../optimizer-cli-reference.md | 7 +-
.../optimizer-configuration.md | 7 +
maintenance/jobs/build.gradle.kts | 7 +
.../TestIcebergRewriteManifestsJobWithSpark.java | 152 ++++++++
.../optimizer/recommender/Recommender.java | 85 +++--
.../handler/ManifestRewriteJobContext.java | 77 +++++
.../handler/ManifestRewriteStrategyHandler.java | 212 ++++++++++++
.../recommender/job/GravitinoJobSubmitter.java | 3 +
.../job/GravitinoManifestRewriteJobAdapter.java | 46 +++
.../TestManifestRewriteRecommendation.java | 334 ++++++++++++++++++
.../recommender/TestRecommenderExecutionMode.java | 60 ++++
.../TestManifestRewriteStrategyHandler.java | 384 +++++++++++++++++++++
.../recommender/job/TestGravitinoJobSubmitter.java | 34 ++
.../server/web/rest/TestPolicyOperations.java | 40 +++
31 files changed, 2184 insertions(+), 79 deletions(-)
diff --git
a/api/src/main/java/org/apache/gravitino/policy/IcebergRewriteManifestsContent.java
b/api/src/main/java/org/apache/gravitino/policy/IcebergRewriteManifestsContent.java
index 53d59202cb..5a00e4e490 100644
---
a/api/src/main/java/org/apache/gravitino/policy/IcebergRewriteManifestsContent.java
+++
b/api/src/main/java/org/apache/gravitino/policy/IcebergRewriteManifestsContent.java
@@ -18,20 +18,18 @@
*/
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 manifest rewrite.
- *
- * <p>Configurable thresholds will be added when the rewrite-manifests policy
lands; this class
- * registers the built-in type for TMS task mapping.
- */
+/** Typed thresholds and target options for Iceberg manifest rewriting. */
public class IcebergRewriteManifestsContent implements PolicyContent {
/** Property key for strategy type. */
public static final String STRATEGY_TYPE_KEY = "strategy.type";
@@ -41,21 +39,109 @@ public class IcebergRewriteManifestsContent 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";
- /** Strategy type for manifest rewrite. */
+ /** Strategy type for manifest rewriting. */
public static final String STRATEGY_TYPE_VALUE = "iceberg-rewrite-manifests";
- /** Built-in job template name for rewrite manifests. */
+ /** Built-in rewrite job template. */
public static final String JOB_TEMPLATE_NAME_VALUE =
"builtin-iceberg-rewrite-manifests";
+ /** Rule key for critical manifest count. */
+ public static final String MANIFEST_COUNT_CRITICAL =
"manifest_count_critical";
+ /** Default critical manifest count. */
+ public static final long DEFAULT_MANIFEST_COUNT_CRITICAL = 500L;
+ /** Rule key for minimum count for the size trigger. */
+ public static final String MANIFEST_COUNT_WARNING = "manifest_count_warning";
+ /** Default minimum count for the size trigger. */
+ public static final long DEFAULT_MANIFEST_COUNT_WARNING = 100L;
+ /** Rule key for exclusive average size threshold in bytes. */
+ public static final String AVG_MANIFEST_SIZE_THRESHOLD_BYTES =
+ "avg_manifest_size_threshold_bytes";
+ /** Default exclusive average size threshold in bytes. */
+ public static final long DEFAULT_AVG_MANIFEST_SIZE_THRESHOLD_BYTES =
8388608L;
+ /** Rule key for the requested partition spec ID; omission selects Iceberg's
current default. */
+ public static final String SPEC_ID = "spec_id";
+ /** Rule key for caching option, or null for the Iceberg default. */
+ public static final String USE_CACHING = "use_caching";
+
+ private final Long manifestCountCritical;
+ private final Long manifestCountWarning;
+ private final Long avgManifestSizeThresholdBytes;
+ private final Integer specId;
+ private final Boolean useCaching;
+
+ /** Creates default thresholds targeting Iceberg's current default partition
spec. */
+ public IcebergRewriteManifestsContent() {
+ this(null, null, null, null, null);
+ }
- private static final Set<MetadataObject.Type> SUPPORTED_OBJECT_TYPES =
- ImmutableSet.of(
- MetadataObject.Type.CATALOG, MetadataObject.Type.SCHEMA,
MetadataObject.Type.TABLE);
+ IcebergRewriteManifestsContent(
+ @Nullable Long manifestCountCritical,
+ @Nullable Long manifestCountWarning,
+ @Nullable Long avgManifestSizeThresholdBytes,
+ @Nullable Integer specId,
+ @Nullable Boolean useCaching) {
+ this.manifestCountCritical =
+ manifestCountCritical == null ? DEFAULT_MANIFEST_COUNT_CRITICAL :
manifestCountCritical;
+ this.manifestCountWarning =
+ manifestCountWarning == null ? DEFAULT_MANIFEST_COUNT_WARNING :
manifestCountWarning;
+ this.avgManifestSizeThresholdBytes =
+ avgManifestSizeThresholdBytes == null
+ ? DEFAULT_AVG_MANIFEST_SIZE_THRESHOLD_BYTES
+ : avgManifestSizeThresholdBytes;
+ this.specId = specId;
+ this.useCaching = useCaching;
+ }
- /** Default constructor for Jackson deserialization only. */
- public IcebergRewriteManifestsContent() {}
+ /**
+ * Returns the critical manifest count.
+ *
+ * @return critical manifest count
+ */
+ public Long manifestCountCritical() {
+ return manifestCountCritical;
+ }
+
+ /**
+ * Returns the minimum count for the size trigger.
+ *
+ * @return minimum count for the size trigger
+ */
+ public Long manifestCountWarning() {
+ return manifestCountWarning;
+ }
+
+ /**
+ * Returns the exclusive average size threshold in bytes.
+ *
+ * @return exclusive average size threshold in bytes
+ */
+ public Long avgManifestSizeThresholdBytes() {
+ return avgManifestSizeThresholdBytes;
+ }
+
+ /**
+ * Returns the requested partition spec ID. When absent, the recommender
resolves Iceberg's
+ * current default from table metadata and retains it through job submission.
+ *
+ * @return requested partition spec ID, or null to use Iceberg's current
default
+ */
+ @Nullable
+ public Integer specId() {
+ return specId;
+ }
+
+ /**
+ * Returns the caching option, or null for the Iceberg default.
+ *
+ * @return caching option, or null for the Iceberg default
+ */
+ @Nullable
+ public Boolean useCaching() {
+ return useCaching;
+ }
@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 +153,50 @@ public class IcebergRewriteManifestsContent 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");
- rules.put(SCORE_EXPR_KEY, "1");
+ rules.put(MANIFEST_COUNT_CRITICAL, manifestCountCritical);
+ rules.put(MANIFEST_COUNT_WARNING, manifestCountWarning);
+ rules.put(AVG_MANIFEST_SIZE_THRESHOLD_BYTES,
avgManifestSizeThresholdBytes);
+ if (specId != null) {
+ rules.put(SPEC_ID, specId);
+ }
+ if (useCaching != null) {
+ rules.put(USE_CACHING, useCaching);
+ }
return Collections.unmodifiableMap(rules);
}
@Override
- public boolean equals(Object other) {
- return other instanceof IcebergRewriteManifestsContent;
+ public void validate() {
+ PolicyContent.super.validate();
+ Preconditions.checkArgument(manifestCountWarning > 0,
"manifest_count_warning must be > 0");
+ Preconditions.checkArgument(
+ manifestCountCritical >= manifestCountWarning,
+ "manifest_count_critical must be >= manifest_count_warning");
+ Preconditions.checkArgument(
+ avgManifestSizeThresholdBytes > 0, "avg_manifest_size_threshold_bytes
must be > 0");
+ Preconditions.checkArgument(specId == null || specId >= 0, "spec_id must
be >= 0");
}
@Override
- public int hashCode() {
- return IcebergRewriteManifestsContent.class.hashCode();
+ public boolean equals(Object other) {
+ if (!(other instanceof IcebergRewriteManifestsContent)) {
+ return false;
+ }
+ IcebergRewriteManifestsContent that = (IcebergRewriteManifestsContent)
other;
+ return Objects.equals(manifestCountCritical, that.manifestCountCritical)
+ && Objects.equals(manifestCountWarning, that.manifestCountWarning)
+ && Objects.equals(avgManifestSizeThresholdBytes,
that.avgManifestSizeThresholdBytes)
+ && Objects.equals(specId, that.specId)
+ && Objects.equals(useCaching, that.useCaching);
}
@Override
- public String toString() {
- return "IcebergRewriteManifestsContent{}";
+ public int hashCode() {
+ return Objects.hash(
+ manifestCountCritical,
+ manifestCountWarning,
+ avgManifestSizeThresholdBytes,
+ specId,
+ useCaching);
}
}
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 9905811353..d1028a29fb 100644
--- a/api/src/main/java/org/apache/gravitino/policy/PolicyContents.java
+++ b/api/src/main/java/org/apache/gravitino/policy/PolicyContents.java
@@ -171,12 +171,38 @@ public class PolicyContents {
}
/**
- * Creates placeholder Iceberg rewrite-manifests policy content.
+ * Creates a manifest rewrite policy with default thresholds targeting
Iceberg's current default
+ * partition spec.
*
- * @return rewrite-manifests policy content
+ * @return manifest rewrite policy
*/
public static PolicyContent icebergRewriteManifests() {
- return new IcebergRewriteManifestsContent();
+ return icebergRewriteManifests(null, null, null, null, null);
+ }
+
+ /**
+ * Creates a manifest rewrite policy. Null thresholds use the built-in
defaults.
+ *
+ * @param manifestCountCritical count that triggers regardless of size
+ * @param manifestCountWarning minimum count for the size trigger
+ * @param avgManifestSizeThresholdBytes exclusive average size threshold in
bytes
+ * @param specId requested existing spec, or null to resolve Iceberg's
current default at
+ * evaluation
+ * @param useCaching caching option, or null for the Iceberg default
+ * @return manifest rewrite policy
+ */
+ public static IcebergRewriteManifestsContent icebergRewriteManifests(
+ @Nullable Long manifestCountCritical,
+ @Nullable Long manifestCountWarning,
+ @Nullable Long avgManifestSizeThresholdBytes,
+ @Nullable Integer specId,
+ @Nullable Boolean useCaching) {
+ return new IcebergRewriteManifestsContent(
+ manifestCountCritical,
+ manifestCountWarning,
+ avgManifestSizeThresholdBytes,
+ specId,
+ useCaching);
}
/**
diff --git
a/api/src/test/java/org/apache/gravitino/policy/TestIcebergRewriteManifestsContent.java
b/api/src/test/java/org/apache/gravitino/policy/TestIcebergRewriteManifestsContent.java
new file mode 100644
index 0000000000..bbdd801abd
--- /dev/null
+++
b/api/src/test/java/org/apache/gravitino/policy/TestIcebergRewriteManifestsContent.java
@@ -0,0 +1,66 @@
+/*
+ * 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 TestIcebergRewriteManifestsContent {
+ @Test
+ void testDefaultsAndRegistration() {
+ IcebergRewriteManifestsContent content = new
IcebergRewriteManifestsContent();
+ content.validate();
+ Assertions.assertEquals(500L, content.manifestCountCritical());
+ Assertions.assertEquals(100L, content.manifestCountWarning());
+ Assertions.assertEquals(8388608L, content.avgManifestSizeThresholdBytes());
+ Assertions.assertNull(content.specId());
+ Assertions.assertNull(content.useCaching());
+ Assertions.assertEquals(content, PolicyContents.icebergRewriteManifests());
+ Assertions.assertEquals(
+ content.hashCode(),
PolicyContents.icebergRewriteManifests().hashCode());
+ Assertions.assertEquals(
+ IcebergRewriteManifestsContent.class,
+
Policy.BuiltInType.fromPolicyType("system_iceberg_rewrite_manifests").contentClass());
+ Assertions.assertThrows(
+ UnsupportedOperationException.class, () ->
content.rules().put("spec_id", 1));
+ }
+
+ @Test
+ void testValidationAndOptions() {
+ for (IcebergRewriteManifestsContent content :
+ new IcebergRewriteManifestsContent[] {
+ PolicyContents.icebergRewriteManifests(99L, 100L, null, null, null),
+ PolicyContents.icebergRewriteManifests(null, 0L, null, null, null),
+ PolicyContents.icebergRewriteManifests(null, null, 0L, null, null),
+ PolicyContents.icebergRewriteManifests(null, null, null, -1, null)
+ }) {
+ Assertions.assertThrows(IllegalArgumentException.class,
content::validate);
+ }
+ IcebergRewriteManifestsContent content =
+ PolicyContents.icebergRewriteManifests(7L, 7L, 1L, 0, false);
+ content.validate();
+ Assertions.assertEquals(0, content.rules().get("spec_id"));
+ Assertions.assertEquals(false, content.rules().get("use_caching"));
+ Assertions.assertNotEquals(content,
PolicyContents.icebergRewriteManifests());
+ IcebergRewriteManifestsContent largestSpec =
+ PolicyContents.icebergRewriteManifests(null, null, null,
Integer.MAX_VALUE, null);
+ largestSpec.validate();
+ Assertions.assertEquals(Integer.MAX_VALUE, largestSpec.specId());
+ }
+}
diff --git
a/catalogs/catalog-common/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergConstants.java
b/catalogs/catalog-common/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergConstants.java
index 120cebb619..8c0e1bb97f 100644
---
a/catalogs/catalog-common/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergConstants.java
+++
b/catalogs/catalog-common/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergConstants.java
@@ -98,6 +98,10 @@ public class IcebergConstants {
public static final String OWNER = "owner";
public static final String LOCATION = "location";
public static final String CURRENT_SNAPSHOT_ID = "current-snapshot-id";
+
+ /** Read-only partition spec ID projected from Iceberg table metadata. */
+ public static final String DEFAULT_SPEC_ID = "default-spec-id";
+
public static final String CHERRY_PICK_SNAPSHOT_ID =
"cherry-pick-snapshot-id";
public static final String SORT_ORDER = "sort-order";
public static final String IDENTIFIER_FIELDS = "identifier-fields";
diff --git
a/catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergTablePropertiesMetadata.java
b/catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergTablePropertiesMetadata.java
index b2eff94881..37a0466afe 100644
---
a/catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergTablePropertiesMetadata.java
+++
b/catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergTablePropertiesMetadata.java
@@ -38,6 +38,10 @@ public class IcebergTablePropertiesMetadata extends
BasePropertiesMetadata {
public static final String CREATOR = IcebergConstants.CREATOR;
public static final String LOCATION = IcebergConstants.LOCATION;
public static final String CURRENT_SNAPSHOT_ID =
IcebergConstants.CURRENT_SNAPSHOT_ID;
+
+ /** Read-only partition spec ID projected from Iceberg table metadata. */
+ public static final String DEFAULT_SPEC_ID =
IcebergConstants.DEFAULT_SPEC_ID;
+
public static final String CHERRY_PICK_SNAPSHOT_ID =
IcebergConstants.CHERRY_PICK_SNAPSHOT_ID;
public static final String SORT_ORDER = IcebergConstants.SORT_ORDER;
public static final String IDENTIFIER_FIELDS =
IcebergConstants.IDENTIFIER_FIELDS;
@@ -77,6 +81,8 @@ public class IcebergTablePropertiesMetadata extends
BasePropertiesMetadata {
CURRENT_SNAPSHOT_ID,
"The snapshot represents the current state of the table",
false),
+ stringReservedPropertyEntry(
+ DEFAULT_SPEC_ID, "The current default partition spec from
Iceberg metadata", false),
stringReservedPropertyEntry(
CHERRY_PICK_SNAPSHOT_ID,
"Selecting a specific snapshot in a merge operation",
diff --git
a/catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/ops/IcebergCatalogWrapperHelper.java
b/catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/ops/IcebergCatalogWrapperHelper.java
index 808a830651..413cd4f4bf 100644
---
a/catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/ops/IcebergCatalogWrapperHelper.java
+++
b/catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/ops/IcebergCatalogWrapperHelper.java
@@ -19,6 +19,8 @@
package org.apache.gravitino.catalog.lakehouse.iceberg.ops;
+import static
org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants.DEFAULT_SPEC_ID;
+
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Joiner;
import com.google.common.base.Preconditions;
@@ -64,6 +66,7 @@ public class IcebergCatalogWrapperHelper {
"location",
"comment",
"current-snapshot-id",
+ DEFAULT_SPEC_ID,
"cherry-pick-snapshot-id",
"sort-order",
"identifier-fields");
diff --git
a/catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/utils/IcebergTablePropertiesUtil.java
b/catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/utils/IcebergTablePropertiesUtil.java
index 75b39ffe08..55ca37154c 100644
---
a/catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/utils/IcebergTablePropertiesUtil.java
+++
b/catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/utils/IcebergTablePropertiesUtil.java
@@ -21,6 +21,7 @@ package org.apache.gravitino.catalog.lakehouse.iceberg.utils;
import static
org.apache.gravitino.catalog.lakehouse.iceberg.IcebergTable.DEFAULT_ICEBERG_PROVIDER;
import static
org.apache.gravitino.catalog.lakehouse.iceberg.IcebergTablePropertiesMetadata.CURRENT_SNAPSHOT_ID;
+import static
org.apache.gravitino.catalog.lakehouse.iceberg.IcebergTablePropertiesMetadata.DEFAULT_SPEC_ID;
import static
org.apache.gravitino.catalog.lakehouse.iceberg.IcebergTablePropertiesMetadata.FORMAT;
import static
org.apache.gravitino.catalog.lakehouse.iceberg.IcebergTablePropertiesMetadata.FORMAT_VERSION;
import static
org.apache.gravitino.catalog.lakehouse.iceberg.IcebergTablePropertiesMetadata.IDENTIFIER_FIELDS;
@@ -54,6 +55,7 @@ public class IcebergTablePropertiesUtil {
? String.valueOf(table.currentSnapshot().snapshotId())
: "none";
properties.put(CURRENT_SNAPSHOT_ID, currentSnapshotId);
+ properties.put(DEFAULT_SPEC_ID, Integer.toString(table.defaultSpecId()));
properties.put(LOCATION, table.location());
properties.put(FORMAT_VERSION, String.valueOf(table.formatVersion()));
diff --git
a/catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/TestIcebergTablePropertiesMetadata.java
b/catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/TestIcebergTablePropertiesMetadata.java
index 08ae281b91..8debcf137a 100644
---
a/catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/TestIcebergTablePropertiesMetadata.java
+++
b/catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/TestIcebergTablePropertiesMetadata.java
@@ -19,6 +19,7 @@
package org.apache.gravitino.catalog.lakehouse.iceberg;
import com.google.common.collect.ImmutableMap;
+import java.util.Map;
import java.util.stream.Stream;
import org.apache.gravitino.catalog.PropertiesMetadataHelpers;
import org.junit.jupiter.api.Assertions;
@@ -68,6 +69,20 @@ public class TestIcebergTablePropertiesMetadata {
Assertions.assertThrows(IllegalArgumentException.class, () ->
validateFormatVersion(value));
}
+ @Test
+ void testDefaultSpecIdCannotBeSetOrRemoved() {
+ Map<String, String> property = Map.of("default-spec-id", "1");
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> PropertiesMetadataHelpers.validatePropertyForCreate(metadata,
property));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> PropertiesMetadataHelpers.validatePropertyForAlter(metadata,
property, Map.of()));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> PropertiesMetadataHelpers.validatePropertyForAlter(metadata,
Map.of(), property));
+ }
+
private static Stream<String> invalidFormatVersions() {
// Just outside the range (0, 5) and clearly out of range / non-numeric.
return Stream.of(
diff --git
a/catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/ops/TestIcebergTableUpdate.java
b/catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/ops/TestIcebergTableUpdate.java
index 2be6a402e2..34cd91a7f7 100644
---
a/catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/ops/TestIcebergTableUpdate.java
+++
b/catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/ops/TestIcebergTableUpdate.java
@@ -163,6 +163,17 @@ public class TestIcebergTableUpdate {
});
}
+ /** Rejects attempts to modify the default spec through ordinary table
properties. */
+ @Test
+ public void testDefaultSpecPropertyIsReadOnly() {
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> updateTable(identifier,
TableChange.setProperty("default-spec-id", "1")));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> updateTable(identifier,
TableChange.removeProperty("default-spec-id")));
+ }
+
@Test
public void testRenameTable() {
TableChange renameTable = TableChange.rename("new_table_name");
diff --git
a/catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/utils/TestIcebergTablePropertiesUtil.java
b/catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/utils/TestIcebergTablePropertiesUtil.java
new file mode 100644
index 0000000000..ebee424b21
--- /dev/null
+++
b/catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/utils/TestIcebergTablePropertiesUtil.java
@@ -0,0 +1,56 @@
+/*
+ * 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.catalog.lakehouse.iceberg.utils;
+
+import java.nio.file.Path;
+import java.util.Map;
+import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergTable;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+class TestIcebergTablePropertiesUtil {
+ @TempDir Path temporaryDirectory;
+
+ @Test
+ void testDefaultSpecComesFromMetadataAndTracksEvolution() {
+ Schema schema = new Schema(Types.NestedField.required(1, "id",
Types.LongType.get()));
+ TableMetadata metadata =
+ TableMetadata.newTableMetadata(
+ schema,
+ PartitionSpec.unpartitioned(),
+ temporaryDirectory.toUri().toString(),
+ Map.of("default-spec-id", "999"));
+ Assertions.assertEquals(0, metadata.defaultSpecId());
+ Assertions.assertEquals(
+ "0", IcebergTable.fromIcebergTable(metadata,
"events").properties().get("default-spec-id"));
+
+ TableMetadata evolved =
+ TableMetadata.buildFrom(metadata)
+
.setDefaultPartitionSpec(PartitionSpec.builderFor(schema).identity("id").build())
+ .build();
+ Assertions.assertEquals(1, evolved.defaultSpecId());
+ Assertions.assertEquals(
+ "1", IcebergTable.fromIcebergTable(evolved,
"events").properties().get("default-spec-id"));
+ }
+}
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 f2417fc4d7..0692a92a24 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
@@ -19,6 +19,12 @@
package org.apache.gravitino.dto.policy;
import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.core.JsonToken;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.JsonDeserializer;
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.LinkedHashMap;
@@ -26,6 +32,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import javax.annotation.Nullable;
+import lombok.AccessLevel;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.EqualsAndHashCode;
@@ -35,6 +42,7 @@ import org.apache.gravitino.authorization.Privilege;
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.PolicyContent;
import org.apache.gravitino.policy.PolicyContents;
import org.apache.gravitino.policy.RowFilterContent;
@@ -307,19 +315,96 @@ public interface PolicyContentDTO extends PolicyContent {
}
}
- /** Placeholder DTO for Iceberg rewrite-manifests policy content. */
+ /** Typed manifest rewrite policy content. */
@EqualsAndHashCode
@ToString
+ @Builder(setterPrefix = "with")
+ @AllArgsConstructor(access = AccessLevel.PRIVATE)
class IcebergRewriteManifestsContentDTO implements PolicyContentDTO {
+ @JsonProperty("manifest_count_critical")
+ @JsonDeserialize(using = ManifestThresholdDeserializer.class)
+ private Long manifestCountCritical;
+
+ @JsonProperty("manifest_count_warning")
+ @JsonDeserialize(using = ManifestThresholdDeserializer.class)
+ private Long manifestCountWarning;
+
+ @JsonProperty("avg_manifest_size_threshold_bytes")
+ @JsonDeserialize(using = ManifestThresholdDeserializer.class)
+ private Long avgManifestSizeThresholdBytes;
+
+ @JsonProperty("spec_id")
+ @JsonDeserialize(using = ManifestSpecIdDeserializer.class)
+ private Integer specId;
+
+ @JsonProperty("use_caching")
+ @JsonDeserialize(using = ManifestUseCachingDeserializer.class)
+ private Boolean useCaching;
+
private IcebergRewriteManifestsContentDTO() {}
/**
- * @return empty rewrite-manifests content DTO
+ * Returns manifest rewrite content with default thresholds.
+ *
+ * @return default rewrite-manifests content DTO
*/
public static IcebergRewriteManifestsContentDTO instance() {
return new IcebergRewriteManifestsContentDTO();
}
+ /**
+ * Returns manifest_count_critical.
+ *
+ * @return manifest_count_critical
+ */
+ public Long manifestCountCritical() {
+ return manifestCountCritical == null
+ ? IcebergRewriteManifestsContent.DEFAULT_MANIFEST_COUNT_CRITICAL
+ : manifestCountCritical;
+ }
+
+ /**
+ * Returns manifest_count_warning.
+ *
+ * @return manifest_count_warning
+ */
+ public Long manifestCountWarning() {
+ return manifestCountWarning == null
+ ? IcebergRewriteManifestsContent.DEFAULT_MANIFEST_COUNT_WARNING
+ : manifestCountWarning;
+ }
+
+ /**
+ * Returns avg_manifest_size_threshold_bytes.
+ *
+ * @return avg_manifest_size_threshold_bytes
+ */
+ public Long avgManifestSizeThresholdBytes() {
+ return avgManifestSizeThresholdBytes == null
+ ?
IcebergRewriteManifestsContent.DEFAULT_AVG_MANIFEST_SIZE_THRESHOLD_BYTES
+ : avgManifestSizeThresholdBytes;
+ }
+
+ /**
+ * Returns spec_id.
+ *
+ * @return spec_id or null when omitted
+ */
+ @Nullable
+ public Integer specId() {
+ return specId;
+ }
+
+ /**
+ * Returns use_caching.
+ *
+ * @return use_caching or null when omitted
+ */
+ @Nullable
+ public Boolean useCaching() {
+ return useCaching;
+ }
+
@Override
public Set<MetadataObject.Type> supportedObjectTypes() {
return toDomainContent().supportedObjectTypes();
@@ -341,8 +426,50 @@ public interface PolicyContentDTO extends PolicyContent {
toDomainContent().validate();
}
+ /** Reads a manifest threshold without silently coercing fractional values
or strings. */
+ public static class ManifestThresholdDeserializer 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, "Manifest thresholds must be 64-bit integers");
+ }
+ return parser.getLongValue();
+ }
+ }
+
+ /** Reads a partition spec ID without silently coercing fractional values
or strings. */
+ public static class ManifestSpecIdDeserializer extends
JsonDeserializer<Integer> {
+ @Override
+ public Integer deserialize(JsonParser parser, DeserializationContext
context)
+ throws IOException {
+ if (!parser.hasToken(JsonToken.VALUE_NUMBER_INT)) {
+ return context.reportInputMismatch(Integer.class, "spec_id must be a
32-bit integer");
+ }
+ return parser.getIntValue();
+ }
+ }
+
+ /** Reads the caching option without silently coercing numbers or strings
to booleans. */
+ public static class ManifestUseCachingDeserializer 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, "use_caching must
be a boolean");
+ }
+ return parser.getBooleanValue();
+ }
+ }
+
private PolicyContent toDomainContent() {
- return PolicyContents.icebergRewriteManifests();
+ return PolicyContents.icebergRewriteManifests(
+ manifestCountCritical(),
+ manifestCountWarning(),
+ avgManifestSizeThresholdBytes(),
+ specId(),
+ useCaching());
}
}
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 0544512fd9..69f7922280 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
@@ -690,7 +690,14 @@ public class DTOConverters {
.build();
}
if (policyContent instanceof IcebergRewriteManifestsContent) {
- return PolicyContentDTO.IcebergRewriteManifestsContentDTO.instance();
+ IcebergRewriteManifestsContent content =
(IcebergRewriteManifestsContent) policyContent;
+ return PolicyContentDTO.IcebergRewriteManifestsContentDTO.builder()
+ .withManifestCountCritical(content.manifestCountCritical())
+ .withManifestCountWarning(content.manifestCountWarning())
+
.withAvgManifestSizeThresholdBytes(content.avgManifestSizeThresholdBytes())
+ .withSpecId(content.specId())
+ .withUseCaching(content.useCaching())
+ .build();
}
if (policyContent instanceof IcebergSnapshotExpirationContent) {
return PolicyContentDTO.IcebergSnapshotExpirationContentDTO.instance();
@@ -1583,7 +1590,14 @@ public class DTOConverters {
content.olderThanDays(), content.location(), content.dryRun());
}
if (policyContentDTO instanceof
PolicyContentDTO.IcebergRewriteManifestsContentDTO) {
- return PolicyContents.icebergRewriteManifests();
+ PolicyContentDTO.IcebergRewriteManifestsContentDTO content =
+ (PolicyContentDTO.IcebergRewriteManifestsContentDTO)
policyContentDTO;
+ return PolicyContents.icebergRewriteManifests(
+ content.manifestCountCritical(),
+ content.manifestCountWarning(),
+ content.avgManifestSizeThresholdBytes(),
+ content.specId(),
+ content.useCaching());
}
if (policyContentDTO instanceof
PolicyContentDTO.IcebergSnapshotExpirationContentDTO) {
return PolicyContents.icebergSnapshotExpiration();
diff --git
a/common/src/test/java/org/apache/gravitino/dto/policy/TestManifestRewritePolicyDTO.java
b/common/src/test/java/org/apache/gravitino/dto/policy/TestManifestRewritePolicyDTO.java
new file mode 100644
index 0000000000..9518c392ad
--- /dev/null
+++
b/common/src/test/java/org/apache/gravitino/dto/policy/TestManifestRewritePolicyDTO.java
@@ -0,0 +1,154 @@
+/*
+ * 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 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.IcebergRewriteManifestsContent;
+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 TestManifestRewritePolicyDTO {
+ @Test
+ void testPolicyAndCreateRequestRoundTrip() throws Exception {
+ IcebergRewriteManifestsContent content =
+ PolicyContents.icebergRewriteManifests(900L, 200L, 4096L, 3, false);
+ String persisted = JsonUtils.anyFieldMapper().writeValueAsString(content);
+ Assertions.assertEquals(
+ content,
+ JsonUtils.anyFieldMapper().readValue(persisted,
IcebergRewriteManifestsContent.class));
+ PolicyContentDTO dto = DTOConverters.toDTO(content);
+ Assertions.assertEquals(content, DTOConverters.fromDTO(dto));
+ String json =
+
"{\"name\":\"rewrite\",\"policyType\":\"system_iceberg_rewrite_manifests\","
+ + "\"enabled\":true,\"content\":"
+ + JsonUtils.objectMapper().writeValueAsString(dto)
+ + "}";
+ PolicyDTO policy = JsonUtils.objectMapper().readValue(json,
PolicyDTO.class);
+ Assertions.assertEquals(content, DTOConverters.fromDTO(policy.content()));
+ PolicyCreateRequest request =
+ JsonUtils.objectMapper().readValue(json, PolicyCreateRequest.class);
+ request.validate();
+ Assertions.assertEquals(content,
DTOConverters.fromDTO(request.getPolicyContent()));
+ }
+
+ @ParameterizedTest
+ @CsvSource({
+ "use_caching,0",
+ "use_caching,1",
+ "use_caching,1.0",
+ "use_caching,\"true\"",
+ "use_caching,\"false\"",
+ "use_caching,[]",
+ "use_caching,{}",
+ "spec_id,1.5",
+ "spec_id,2147483648",
+ "spec_id,true",
+ "spec_id,\"1\"",
+ "manifest_count_critical,500.5",
+ "manifest_count_critical,9223372036854775808",
+ "manifest_count_warning,100.5",
+ "manifest_count_warning,false",
+ "avg_manifest_size_threshold_bytes,4096.5",
+ "avg_manifest_size_threshold_bytes,\"4096\""
+ })
+ void testRejectInvalidValuesInCreateAndUpdate(String field, String value) {
+ String content = "{\"" + field + "\":" + value + "}";
+ String create =
+
"{\"name\":\"rewrite\",\"policyType\":\"system_iceberg_rewrite_manifests\","
+ + "\"enabled\":true,\"content\":"
+ + content
+ + "}";
+ String update =
+
"{\"@type\":\"updateContent\",\"policyType\":\"system_iceberg_rewrite_manifests\","
+ + "\"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 testBooleanCachingValues(boolean useCaching) throws Exception {
+ PolicyContentDTO.IcebergRewriteManifestsContentDTO dto =
+ JsonUtils.objectMapper()
+ .readValue(
+ "{\"use_caching\":" + useCaching + "}",
+ PolicyContentDTO.IcebergRewriteManifestsContentDTO.class);
+ dto.validate();
+ Assertions.assertEquals(
+ PolicyContents.icebergRewriteManifests(null, null, null, null,
useCaching),
+ DTOConverters.fromDTO(dto));
+ }
+
+ @Test
+ void testNumericBoundsAndNullDefaults() throws Exception {
+ PolicyContentDTO.IcebergRewriteManifestsContentDTO bounds =
+ JsonUtils.objectMapper()
+ .readValue(
+
"{\"spec_id\":2147483647,\"manifest_count_critical\":9223372036854775807}",
+ PolicyContentDTO.IcebergRewriteManifestsContentDTO.class);
+ bounds.validate();
+ Assertions.assertEquals(Integer.MAX_VALUE, bounds.specId());
+ Assertions.assertEquals(Long.MAX_VALUE, bounds.manifestCountCritical());
+ PolicyContentDTO.IcebergRewriteManifestsContentDTO defaults =
+ JsonUtils.objectMapper()
+ .readValue(
+
"{\"spec_id\":null,\"manifest_count_critical\":null,\"manifest_count_warning\":null,"
+ +
"\"avg_manifest_size_threshold_bytes\":null,\"use_caching\":null}",
+ PolicyContentDTO.IcebergRewriteManifestsContentDTO.class);
+ defaults.validate();
+ Assertions.assertEquals(
+ PolicyContents.icebergRewriteManifests(),
DTOConverters.fromDTO(defaults));
+ }
+
+ @Test
+ void testUpdateAndOmittedDefaults() throws Exception {
+ String json =
+
"{\"@type\":\"updateContent\",\"policyType\":\"system_iceberg_rewrite_manifests\",\"newContent\":{}}";
+ PolicyUpdateRequest.UpdatePolicyContentRequest request =
+ JsonUtils.objectMapper()
+ .readValue(json,
PolicyUpdateRequest.UpdatePolicyContentRequest.class);
+ request.validate();
+ PolicyContentDTO.IcebergRewriteManifestsContentDTO dto =
+ JsonUtils.objectMapper()
+ .readValue("{}",
PolicyContentDTO.IcebergRewriteManifestsContentDTO.class);
+ Assertions.assertEquals(PolicyContents.icebergRewriteManifests(),
DTOConverters.fromDTO(dto));
+ dto.validate();
+ PolicyContentDTO.IcebergRewriteManifestsContentDTO invalid =
+ JsonUtils.objectMapper()
+ .readValue(
+ "{\"spec_id\":-1}",
PolicyContentDTO.IcebergRewriteManifestsContentDTO.class);
+ Assertions.assertThrows(IllegalArgumentException.class, invalid::validate);
+ }
+}
diff --git a/docs/iceberg-rewrite-manifests-policy.md
b/docs/iceberg-rewrite-manifests-policy.md
new file mode 100644
index 0000000000..53142db2b9
--- /dev/null
+++ b/docs/iceberg-rewrite-manifests-policy.md
@@ -0,0 +1,105 @@
+---
+title: "Iceberg Manifest Rewrite Policy"
+slug: "/iceberg-rewrite-manifests-policy"
+keyword: "iceberg, manifests, policy, optimizer, Gravitino"
+license: "This software is licensed under the Apache License version 2."
+---
+
+## Behavior
+
+`system_iceberg_rewrite_manifests` evaluates manifest statistics for one
resolved partition
+spec and submits `builtin-iceberg-rewrite-manifests`. Associate the policy
with a tag and
+assign that tag to a catalog, schema, or table through the existing
+[policy API](./manage-policies-in-gravitino.md). Tables inherit tags from
their ancestors.
+The strategy type is `iceberg-rewrite-manifests`; its handler and job adapter
are built in.
+
+| Content field | Default | Meaning |
+| --- | --- | --- |
+| `manifest_count_critical` | `500` | Trigger when count is at least this
value, regardless of size. |
+| `manifest_count_warning` | `100` | Minimum count for the small-average-size
trigger. Must be positive. |
+| `avg_manifest_size_threshold_bytes` | `8388608` | Trigger when average size
is strictly below this value and the warning count is reached. Must be
positive. |
+| `spec_id` | Iceberg's current default | Existing partition spec to evaluate.
Omission resolves the current default from Iceberg metadata. |
+| `use_caching` | Omitted | Optional boolean forwarded to Iceberg; omission
preserves the runtime default. |
+
+The critical count must be at least the warning count. Spec IDs must be
non-negative
+32-bit integers. Thresholds use integer counts and bytes. Recommendations are
ranked by
+the selected spec's manifest count. The policy API accepts only JSON booleans
for
+`use_caching`; numbers and strings are rejected.
+
+At the defaults, 99 manifests never trigger; 100 manifests averaging 8 MiB do
not trigger;
+100 averaging less than 8 MiB trigger; and 500 trigger regardless of size.
+
+## Collect, evaluate, and submit
+
+Submit `builtin-iceberg-update-manifest-stats` to collect complete
+`custom-manifest-number-by-spec` and `custom-avg-manifest-size-by-spec` maps
plus
+`custom-manifest-snapshot-id` from one table snapshot. The collector covers
every defined
+spec in one traversal; it has no `spec_id` argument. Both maps are keyed by
decimal spec ID.
+
+The handler compares the published snapshot ID with `current-snapshot-id` in
the table
+metadata supplied by the optimizer's table metadata provider. Missing,
malformed, or
+mismatched snapshot IDs produce no recommendation; recollect before evaluating
again.
+For a table without a snapshot, the collector publishes `-1` and Iceberg table
metadata
+reports `none`. The provider must expose this metadata property to verify
freshness.
+The recommender reads metadata after statistics to detect snapshot changes
during collection
+of its inputs. This freshness check is not transactional with job submission;
the table may
+still change after evaluation.
+
+The handler reads the chosen spec's count and average from the same statistics
response.
+A missing map or key produces no recommendation. An empty spec has count and
average zero.
+Invalid policy rules or malformed measurements are logged and skipped for that
table without
+aborting recommendations for other tables. Counts from other specs are ignored.
+
+The recommender and CLI use the table's current default partition spec when
`spec_id` is
+omitted. The following example uses default thresholds and assumes statistics
have been collected:
+
+```shell
+curl -X POST -H 'Content-Type: application/json' \
+ -d '{
+ "name": "rewrite_manifests",
+ "policyType": "system_iceberg_rewrite_manifests",
+ "enabled": true,
+ "content": {}
+ }' \
+ http://localhost:8090/api/metalakes/test/policies
+```
+
+Associate `rewrite_manifests` with a tag using the `ALL_VALUES` selector, then
assign that tag
+to the table. Configure the optimizer's Gravitino providers and Spark
submission settings as described in
+[Optimizer
configuration](./table-maintenance-service/optimizer-configuration.md), then
preview:
+
+```shell
+./bin/gravitino-optimizer.sh --type submit-strategy-jobs \
+ --identifiers rest_catalog.db.t1 --strategy-name rewrite_manifests --dry-run
+```
+
+Remove `--dry-run` to submit the recommendations. The adapter supplies
`catalog_name`,
+`table_identifier`, the exact evaluated `spec_id`, and optional `use_caching`.
+Its catalog, table, spec, and caching values override shared job-submitter
configuration.
+When the policy omits `use_caching`, the adapter supplies an empty option to
clear any shared
+caching setting and leave the Iceberg runtime default in effect.
+Recollect statistics after a successful rewrite before evaluating again.
+
+## Selecting a partition spec
+
+When `spec_id` is omitted, the recommender resolves Iceberg's current
`default-spec-id`
+from the table metadata supplied by the metadata provider. Empty policy
content therefore
+evaluates the current default spec with the default thresholds. Set `spec_id`
explicitly to
+target another existing spec, such as an older spec after partition evolution.
+
+The Iceberg catalog projects `TableMetadata.defaultSpecId()` as the reserved,
read-only table
+property `default-spec-id`. This is metadata, not an operator setting; it
cannot be set or
+removed through Gravitino table-property changes. Custom metadata providers
must expose the
+same value from Iceberg metadata. If an omitted target cannot be resolved
because metadata is
+missing or invalid, the recommender warns and skips that table without
aborting the batch.
+
+The resolved ID is used for both statistics lookup and job submission.
Partition evolution
+after evaluation cannot retarget an already-created job context. A later
evaluation with an
+omitted `spec_id` resolves the new default; an explicit policy target remains
unchanged.
+The handler never infers a spec from statistics map keys. After a snapshot
change, recollect
+statistics before evaluating again.
+
+Rewriting reorganizes manifests within the selected spec and does not migrate
data files
+between specs. This policy introduces no scheduling, cooldown, or last-success
statistic.
+For runtime requirements and direct job submission, see the
+[rewrite-manifests job
reference](./table-maintenance-service/optimizer-cli-reference.md).
diff --git a/docs/lakehouse-iceberg-catalog.md
b/docs/lakehouse-iceberg-catalog.md
index 9b57d22a74..4a7c410b95 100644
--- a/docs/lakehouse-iceberg-catalog.md
+++ b/docs/lakehouse-iceberg-catalog.md
@@ -473,6 +473,7 @@ Pass [Iceberg table
properties](https://iceberg.apache.org/docs/1.5.2/configurat
| `comment` | The table comment; use the `comment` field in
table meta instead.
| (none)
| No | Yes | No |
| `creator` | The table creator.
| (none)
| No | Yes | No |
| `current-snapshot-id` | The snapshot represents the current state of the
table.
| (none)
| No | Yes | No |
+| `default-spec-id` | The current default partition spec ID projected
from Iceberg metadata. Read-only; cannot be set or removed.
| (none)
| No | Yes | No |
| `cherry-pick-snapshot-id` | Selecting a specific snapshot in a merge
operation.
|
(none) | No | Yes | No |
| `sort-order` | Iceberg table sort order; use `SortOrder` in
table meta instead.
| (none)
| No | Yes | No |
| `identifier-fields` | The identifier fields for defining the table.
| (none)
| No | Yes | No |
diff --git a/docs/open-api/policies.yaml b/docs/open-api/policies.yaml
index 8f8a9f1a61..781080ff47 100644
--- a/docs/open-api/policies.yaml
+++ b/docs/open-api/policies.yaml
@@ -459,8 +459,37 @@ components:
IcebergRewriteManifestsContent:
type: object
- description: Placeholder content for system_iceberg_rewrite_manifests.
Configurable thresholds land in a follow-up.
+ description: Thresholds for one resolved Iceberg partition spec.
Supports catalog, schema, and table policies.
additionalProperties: false
+ properties:
+ manifest_count_critical:
+ type: integer
+ format: int64
+ minimum: 1
+ default: 500
+ description: Inclusive count trigger, regardless of size. Must be at
least manifest_count_warning.
+ manifest_count_warning:
+ type: integer
+ format: int64
+ minimum: 1
+ default: 100
+ description: Inclusive minimum count for the average-size trigger.
+ avg_manifest_size_threshold_bytes:
+ type: integer
+ format: int64
+ minimum: 1
+ default: 8388608
+ description: Exclusive average manifest size threshold in bytes.
+ spec_id:
+ type: integer
+ format: int32
+ minimum: 0
+ nullable: true
+ description: Existing partition spec. Omission resolves Iceberg's
current default spec from table metadata. If it cannot be resolved, the table
is skipped. The evaluated ID is retained through job submission.
+ use_caching:
+ type: boolean
+ nullable: true
+ description: Optional caching setting. Omission preserves the
Iceberg runtime default.
IcebergRewriteManifestsPolicy:
allOf:
diff --git a/docs/policies.md b/docs/policies.md
index 3e3889c075..c1ffba9dbc 100644
--- a/docs/policies.md
+++ b/docs/policies.md
@@ -35,14 +35,15 @@ policy. A table with only `data_domain=risk` does not.
### Policy Types and Content
-| Type | Rules
| Consumer |
-|--------------------------------------|--------------------------------------|---------------------------|
-| `system_iceberg_compaction` | Compaction thresholds and scheduling
| Table maintenance service |
-| `system_iceberg_orphan_file_removal` | Orphan-file cleanup options
| Table maintenance service |
-| `system_row_filter` | Row-filter expression
| Not enforced yet |
-| `system_column_mask` | Column-mask expression
| Not enforced yet |
-| `system_access_control` | Privileges and applicable roles
| Not enforced yet |
-| `custom` | A free-form map that you define
| A system that you provide |
+| Type | Rules
| Consumer |
+| ------------------------------------ |
------------------------------------------ | ------------------------- |
+| `system_iceberg_compaction` | Compaction thresholds and scheduling
| Table maintenance service |
+| `system_iceberg_rewrite_manifests` | Manifest count and average-size
thresholds | Table maintenance service |
+| `system_iceberg_orphan_file_removal` | Orphan-file cleanup options
| Table maintenance service |
+| `system_row_filter` | Row-filter expression
| Not enforced yet |
+| `system_column_mask` | Column-mask expression
| Not enforced yet |
+| `system_access_control` | Privileges and applicable roles
| Not enforced yet |
+| `custom` | A free-form map that you define
| A system that you provide |
The row-filter, column-mask, and access-control policy types can be created
and associated with
tags, but Gravitino does not enforce them yet. Until enforcement is available,
these policy types
@@ -170,7 +171,8 @@ do not re-parse stored policy expressions.
A custom policy's rules live in `customRules`. Gravitino stores them and
returns them to clients;
it does not interpret their names or values. Built-in types have a defined
content shape. See
-[Iceberg compaction policy](./iceberg-compaction-policy.md) for the compaction
rules and
+[Iceberg compaction policy](./iceberg-compaction-policy.md) for the compaction
rules,
+[Iceberg manifest rewrite policy](./iceberg-rewrite-manifests-policy.md) for
manifest thresholds, and
[Table maintenance service](./table-maintenance-service/optimizer.md) for a
worked example.
Policy content also has `properties` and `supportedObjectTypes`. Properties
describe the policy
diff --git a/docs/table-maintenance-service/optimizer-cli-reference.md
b/docs/table-maintenance-service/optimizer-cli-reference.md
index 90614f575b..f01ace00cc 100644
--- a/docs/table-maintenance-service/optimizer-cli-reference.md
+++ b/docs/table-maintenance-service/optimizer-cli-reference.md
@@ -509,7 +509,7 @@ Leave `use_caching` unset to use the installed Iceberg
version's default (`false
### How to find `spec_id`
-Omit `spec_id` for routine maintenance of the current partition spec. Do not
guess IDs such as `0` or `1`. Read `default-spec-id` and the `partition-specs`
array from the table's current Iceberg metadata JSON. The array maps each
`spec-id` to its partition fields and transforms; `default-spec-id` identifies
the current spec. These are Iceberg table metadata fields, not Gravitino
catalog properties.
+Omit `spec_id` for routine maintenance of the current partition spec. Do not
guess IDs such as `0` or `1`. Read the reserved, read-only Gravitino table
property `default-spec-id` for the current default. Use the `partition-specs`
array in the table's Iceberg metadata JSON to see each `spec-id`'s partition
fields and transforms.
To discover which specs have manifests in the current snapshot, run:
@@ -518,7 +518,7 @@ SELECT DISTINCT partition_spec_id
FROM rest_catalog.db.t1.manifests;
```
-This query lists represented specs, not which one is current. A defined spec
with no manifests may be absent; use the metadata JSON to identify the default
and interpret the transforms.
+This query lists represented specs, not which one is current. A defined spec
with no manifests may be absent; use `default-spec-id` to identify the current
default and the metadata JSON to interpret the transforms.
For example, suppose metadata shows spec `0` uses `day(event_time)` and the
current spec `1` uses `hour(event_time)`. Omitting `spec_id` rewrites eligible
manifests for spec `1`. Passing `"spec_id": "0"` consolidates the old day-spec
manifests. Neither run converts day-partitioned data files to hour partitioning
or rewrites manifests belonging to the other spec.
@@ -572,6 +572,9 @@ Rewrite Manifests Results: Rewritten manifests: 24, Added
manifests: 2
Both counts at zero indicate a successful no-op, for example when no manifests
for the selected spec need rewriting. Other specs remain unchanged.
+The [manifest rewrite policy](../iceberg-rewrite-manifests-policy.md)
evaluates fresh per-spec
+measurements from the table-wide collector and submits this job with the
evaluated `spec_id`.
+
## Related
- [Table Maintenance Service](./optimizer.md) for the concepts and the
walkthrough
diff --git a/docs/table-maintenance-service/optimizer-configuration.md
b/docs/table-maintenance-service/optimizer-configuration.md
index fe10b1321b..ffa9334358 100644
--- a/docs/table-maintenance-service/optimizer-configuration.md
+++ b/docs/table-maintenance-service/optimizer-configuration.md
@@ -53,6 +53,13 @@ gravitino.optimizer.jobSubmitterConfig.warehouse_location =
gravitino.optimizer.jobSubmitterConfig.spark_conf =
{"spark.master":"local[2]","spark.hadoop.fs.defaultFS":"file:///"}
```
+For manifest rewrite policies, the mapping
+`gravitino.optimizer.strategyHandler.iceberg-rewrite-manifests.className` is
optional. When
+omitted or blank, the recommender uses the built-in
+`org.apache.gravitino.maintenance.optimizer.recommender.handler.ManifestRewriteStrategyHandler`.
+An explicit class name overrides that default. An omitted policy `spec_id`
uses Iceberg's
+current default spec from table metadata; see [Iceberg manifest rewrite
policy](../iceberg-rewrite-manifests-policy.md).
+
When the Gravitino server has authentication enabled,
`builtin-iceberg-update-stats` needs
credentials in `--updater-options` / `updater_options` for the Gravitino
client (statistics
updater and `submit-update-stats-job` `runJob` calls):
diff --git a/maintenance/jobs/build.gradle.kts
b/maintenance/jobs/build.gradle.kts
index ae5055bda8..2d40e6a663 100644
--- a/maintenance/jobs/build.gradle.kts
+++ b/maintenance/jobs/build.gradle.kts
@@ -54,6 +54,7 @@ dependencies {
}
testImplementation(project(":api"))
+ testImplementation(project(":maintenance:optimizer"))
testImplementation(project(":common"))
// Used to check that the placeholders of the built-in templates resolve the
way the server does.
testImplementation(project(":core"))
@@ -83,6 +84,12 @@ dependencies {
testRuntimeOnly(libs.junit.jupiter.engine)
}
+// The optimizer packages its runtime dependencies beside its jar in
build/libs.
+// Order the integration-test compile after that directory has been populated.
+tasks.compileTestJava {
+ dependsOn(":maintenance:optimizer:copyDepends")
+}
+
tasks.test {
useJUnitPlatform()
}
diff --git
a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteManifestsJobWithSpark.java
b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteManifestsJobWithSpark.java
index e518511c6e..e91877dcea 100644
---
a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteManifestsJobWithSpark.java
+++
b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteManifestsJobWithSpark.java
@@ -26,12 +26,28 @@ import java.io.ByteArrayOutputStream;
import java.io.PrintStream;
import java.nio.charset.StandardCharsets;
import java.nio.file.Path;
+import java.time.Instant;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.dto.AuditDTO;
+import org.apache.gravitino.dto.policy.PolicyDTO;
+import org.apache.gravitino.dto.rel.TableDTO;
+import org.apache.gravitino.dto.util.DTOConverters;
import org.apache.gravitino.maintenance.jobs.TemplateArguments;
+import org.apache.gravitino.maintenance.optimizer.api.common.StatisticEntry;
+import
org.apache.gravitino.maintenance.optimizer.api.recommender.StrategyHandlerContext;
+import
org.apache.gravitino.maintenance.optimizer.common.IcebergManifestStatistics;
+import
org.apache.gravitino.maintenance.optimizer.recommender.handler.ManifestRewriteStrategyHandler;
+import
org.apache.gravitino.maintenance.optimizer.recommender.job.GravitinoManifestRewriteJobAdapter;
+import
org.apache.gravitino.maintenance.optimizer.recommender.strategy.GravitinoStrategy;
+import org.apache.gravitino.policy.PolicyContents;
+import org.apache.iceberg.BaseTable;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.spark.Spark3Util;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.junit.jupiter.api.AfterEach;
@@ -140,6 +156,142 @@ public class TestIcebergRewriteManifestsJobWithSpark {
assertEquals(6, records().size());
}
+ /**
+ * Recollects stale measurements without changing the target selected before
partition evolution.
+ *
+ * @throws Exception if the Iceberg table cannot be loaded
+ */
+ @Test
+ public void testCollectEvaluateAndRewriteAfterDefaultSpecChanges() throws
Exception {
+ int resolvedSpecId = Spark3Util.loadIcebergTable(spark,
TABLE).spec().specId();
+ PolicyDTO policy =
+ PolicyDTO.builder()
+ .withName("rewrite")
+ .withPolicyType("system_iceberg_rewrite_manifests")
+ .withContent(
+ DTOConverters.toDTO(
+ PolicyContents.icebergRewriteManifests(3L, 3L, null,
resolvedSpecId, false)))
+ .build();
+ List<StatisticEntry<?>> measurements =
+ IcebergUpdateManifestStatsJob.collectManifestStatistics(
+ spark, "manifest_catalog", "db.events");
+ IcebergManifestStatistics measurement =
+ IcebergManifestStatistics.fromStatistics(
+ measurements.stream()
+ .collect(Collectors.toMap(StatisticEntry::name,
StatisticEntry::value)),
+ resolvedSpecId)
+ .orElseThrow();
+ assertEquals(1, resolvedSpecId);
+ assertEquals(3, measurement.count());
+ // Change the default after collection, before evaluation and submission.
+ spark.sql(
+ "ALTER TABLE "
+ + TABLE
+ + " REPLACE PARTITION FIELD hours(event_time) WITH
months(event_time)");
+ insertRows(6);
+ List<Row> records = records();
+ Set<String> oldManifests = manifests(0);
+ Set<String> newDefaultManifests = manifests(2);
+ ManifestRewriteStrategyHandler handler = new
ManifestRewriteStrategyHandler();
+ handler.initialize(manifestContext(policy, measurements));
+ assertFalse(handler.shouldTrigger(), "Measurements from an older snapshot
must be recollected");
+ measurements =
+ IcebergUpdateManifestStatsJob.collectManifestStatistics(
+ spark, "manifest_catalog", "db.events");
+ handler.initialize(manifestContext(policy, measurements));
+ assertTrue(handler.shouldTrigger());
+ Map<String, String> config =
+ new GravitinoManifestRewriteJobAdapter()
+ .jobConfig(handler.evaluate().jobExecutionContext().orElseThrow());
+ assertEquals("1", config.get("spec_id"));
+ assertEquals("manifest_catalog", config.get("catalog_name"));
+ assertEquals("db.events", config.get("table_identifier"));
+ runJob(config.get("spec_id"), config.get("use_caching"));
+ assertEquals(1, manifests(1).size());
+ assertEquals(oldManifests, manifests(0));
+ assertEquals(newDefaultManifests, manifests(2));
+ assertEquals(records, records());
+ handler.initialize(
+ manifestContext(
+ policy,
+ IcebergUpdateManifestStatsJob.collectManifestStatistics(
+ spark, "manifest_catalog", "db.events")));
+ assertFalse(handler.shouldTrigger());
+ }
+
+ /**
+ * Resolves an omitted spec from Iceberg metadata and retains it through
later partition
+ * evolution.
+ *
+ * @throws Exception if the Iceberg table cannot be loaded
+ */
+ @Test
+ public void testDefaultSpecIsPinnedAcrossPartitionEvolution() throws
Exception {
+ PolicyDTO policy =
+ PolicyDTO.builder()
+ .withName("rewrite")
+ .withPolicyType("system_iceberg_rewrite_manifests")
+ .withContent(
+ DTOConverters.toDTO(
+ PolicyContents.icebergRewriteManifests(3L, 3L, null, null,
false)))
+ .build();
+ ManifestRewriteStrategyHandler handler = new
ManifestRewriteStrategyHandler();
+ handler.initialize(
+ manifestContext(
+ policy,
+ IcebergUpdateManifestStatsJob.collectManifestStatistics(
+ spark, "manifest_catalog", "db.events")));
+ assertTrue(handler.shouldTrigger(), "An omitted spec must use the current
Iceberg default");
+ Map<String, String> config =
+ new GravitinoManifestRewriteJobAdapter()
+ .jobConfig(handler.evaluate().jobExecutionContext().orElseThrow());
+ assertEquals("1", config.get("spec_id"));
+
+ spark.sql(
+ "ALTER TABLE "
+ + TABLE
+ + " REPLACE PARTITION FIELD hours(event_time) WITH
months(event_time)");
+ insertRows(6);
+ Set<String> oldManifests = manifests(0);
+ Set<String> newDefaultManifests = manifests(2);
+ List<Row> records = records();
+ runJob(config.get("spec_id"), config.get("use_caching"));
+ assertEquals(1, manifests(1).size());
+ assertEquals(oldManifests, manifests(0));
+ assertEquals(newDefaultManifests, manifests(2));
+ assertEquals(records, records());
+
+ handler.initialize(
+ manifestContext(
+ policy,
+ IcebergUpdateManifestStatsJob.collectManifestStatistics(
+ spark, "manifest_catalog", "db.events")));
+ assertTrue(handler.shouldTrigger());
+ assertEquals(
+ "2",
handler.evaluate().jobExecutionContext().orElseThrow().jobOptions().get("spec_id"));
+ }
+
+ private StrategyHandlerContext manifestContext(
+ PolicyDTO policy, List<StatisticEntry<?>> measurements) throws Exception
{
+ TableMetadata metadata =
+ ((BaseTable) Spark3Util.loadIcebergTable(spark,
TABLE)).operations().current();
+ return StrategyHandlerContext.builder(
+ NameIdentifier.of("manifest_catalog", "db", "events"), new
GravitinoStrategy(policy))
+ .withTableStatistics(measurements)
+ .withTableMetadata(
+ TableDTO.builder()
+ .withName("events")
+ .withAudit(
+
AuditDTO.builder().withCreator("test").withCreateTime(Instant.EPOCH).build())
+ .withProperties(
+ Map.of(
+ "current-snapshot-id",
+
Long.toString(metadata.currentSnapshot().snapshotId()),
+ "default-spec-id",
Integer.toString(metadata.defaultSpecId())))
+ .build())
+ .build();
+ }
+
private void runJob(String spec, String caching) {
spark.stop();
SparkSession.clearActiveSession();
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 8da6a52cf3..d73ceb4aa0 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
@@ -49,6 +49,8 @@ import
org.apache.gravitino.maintenance.optimizer.common.CloseableGroup;
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.policy.IcebergRewriteManifestsContent;
import org.apache.gravitino.rel.Table;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -264,29 +266,40 @@ public class Recommender implements AutoCloseable {
PriorityQueue<StrategyEvaluation> scoreQueue = new
PriorityQueue<>(evaluationComparator());
for (NameIdentifier identifier : identifiers) {
- StrategyHandler strategyHandler = loadStrategyHandler(strategy,
identifier);
- if (!strategyHandler.shouldTrigger()) {
+ StrategyHandler strategyHandler =
createStrategyHandler(strategy.strategyType());
+ StrategyHandlerContext context =
+ loadStrategyHandlerContext(strategy, identifier,
strategyHandler.dataRequirements());
+ try {
+ strategyHandler.initialize(context);
+ if (!strategyHandler.shouldTrigger()) {
+ LOG.info(
+ "Skip strategy {} for identifier {} because strategy handler
trigger condition is not met",
+ strategyName,
+ identifier);
+ continue;
+ }
+ StrategyEvaluation evaluation = strategyHandler.evaluate();
+ if (evaluation.score() < 0 ||
evaluation.jobExecutionContext().isEmpty()) {
+ LOG.info(
+ "Skip strategy {} for identifier {} because evaluation score is
negative "
+ + "or job execution context is missing",
+ strategyName,
+ identifier);
+ continue;
+ }
LOG.info(
- "Skip strategy {} for identifier {} because strategy handler
trigger condition is not met",
+ "Recommend strategy {} for identifier {} score: {}",
strategyName,
- identifier);
- continue;
- }
- StrategyEvaluation evaluation = strategyHandler.evaluate();
- if (evaluation.score() < 0 ||
evaluation.jobExecutionContext().isEmpty()) {
- LOG.info(
- "Skip strategy {} for identifier {} because evaluation score is
negative "
- + "or job execution context is missing",
+ identifier,
+ evaluation.score());
+ scoreQueue.add(evaluation);
+ } catch (IllegalArgumentException e) {
+ LOG.warn(
+ "Skip strategy {} for identifier {} because policy rules or
statistics are invalid",
strategyName,
- identifier);
- continue;
+ identifier,
+ e);
}
- LOG.info(
- "Recommend strategy {} for identifier {} score: {}",
- strategyName,
- identifier,
- evaluation.score());
- scoreQueue.add(evaluation);
}
List<StrategyEvaluation> results = new ArrayList<>();
@@ -307,10 +320,8 @@ public class Recommender implements AutoCloseable {
.orElse(""));
}
- private StrategyHandler loadStrategyHandler(Strategy strategy,
NameIdentifier nameIdentifier) {
- StrategyHandler strategyHandler =
createStrategyHandler(strategy.strategyType());
-
- Set<DataRequirement> declaredRequirements =
strategyHandler.dataRequirements();
+ private StrategyHandlerContext loadStrategyHandlerContext(
+ Strategy strategy, NameIdentifier nameIdentifier, Set<DataRequirement>
declaredRequirements) {
EnumSet<DataRequirement> requirements =
declaredRequirements.isEmpty()
? EnumSet.noneOf(DataRequirement.class)
@@ -319,10 +330,6 @@ public class Recommender implements AutoCloseable {
List<StatisticEntry<?>> statistics = Collections.emptyList();
Map<PartitionPath, List<StatisticEntry<?>>> partitionStatistics =
Collections.emptyMap();
- if (requirements.contains(DataRequirement.TABLE_METADATA)) {
- tableMetadata = tableMetadataProvider.tableMetadata(nameIdentifier);
- }
-
if (requirements.contains(DataRequirement.TABLE_STATISTICS)
|| requirements.contains(DataRequirement.PARTITION_STATISTICS)) {
SupportTableStatistics supportTableStatistics =
requireTableStatisticsProvider();
@@ -334,15 +341,16 @@ public class Recommender implements AutoCloseable {
}
}
- StrategyHandlerContext context =
- StrategyHandlerContext.builder(nameIdentifier, strategy)
- .withTableMetadata(tableMetadata)
- .withTableStatistics(statistics)
- .withPartitionStatistics(partitionStatistics)
- .build();
- strategyHandler.initialize(context);
+ // Read metadata after statistics so the freshness check sees intervening
snapshot changes.
+ if (requirements.contains(DataRequirement.TABLE_METADATA)) {
+ tableMetadata = tableMetadataProvider.tableMetadata(nameIdentifier);
+ }
- return strategyHandler;
+ return StrategyHandlerContext.builder(nameIdentifier, strategy)
+ .withTableMetadata(tableMetadata)
+ .withTableStatistics(statistics)
+ .withPartitionStatistics(partitionStatistics)
+ .build();
}
private StrategyHandler createStrategyHandler(String strategyType) {
@@ -371,7 +379,12 @@ public class Recommender implements AutoCloseable {
* {@code COMPACTION}) to {@link StrategyHandler} implementations.
*/
private String getStrategyHandlerClassName(String strategyType) {
- return optimizerEnv.config().getStrategyHandlerClassName(strategyType);
+ String configured =
optimizerEnv.config().getStrategyHandlerClassName(strategyType);
+ if (StringUtils.isBlank(configured)
+ &&
IcebergRewriteManifestsContent.STRATEGY_TYPE_VALUE.equals(strategyType)) {
+ return ManifestRewriteStrategyHandler.class.getName();
+ }
+ return configured;
}
private Map<String, List<NameIdentifier>> getIdentifiersByStrategyName(
diff --git
a/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/ManifestRewriteJobContext.java
b/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/ManifestRewriteJobContext.java
new file mode 100644
index 0000000000..74f2dbb334
--- /dev/null
+++
b/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/ManifestRewriteJobContext.java
@@ -0,0 +1,77 @@
+/*
+ * 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;
+
+import com.google.common.base.Preconditions;
+import java.util.Map;
+import java.util.Objects;
+import javax.annotation.Nullable;
+import org.apache.gravitino.NameIdentifier;
+import
org.apache.gravitino.maintenance.optimizer.api.recommender.JobExecutionContext;
+import org.apache.gravitino.policy.IcebergRewriteManifestsContent;
+
+/** Immutable rewrite request retaining the spec whose statistics were
evaluated. */
+public final class ManifestRewriteJobContext implements JobExecutionContext {
+ private final NameIdentifier name;
+ private final int specId;
+ @Nullable private final Boolean useCaching;
+
+ /**
+ * Creates a request for one resolved partition spec.
+ *
+ * @param name normalized catalog/schema/table identifier
+ * @param specId resolved existing partition spec
+ * @param useCaching optional Iceberg caching setting
+ */
+ public ManifestRewriteJobContext(NameIdentifier name, int specId, @Nullable
Boolean useCaching) {
+ Preconditions.checkArgument(specId >= 0, "spec_id must be >= 0");
+ this.name = Objects.requireNonNull(name, "name");
+ this.specId = specId;
+ this.useCaching = useCaching;
+ }
+
+ /**
+ * Returns the resolved spec, never the table's subsequently changed default.
+ *
+ * @return resolved partition spec ID
+ */
+ public int specId() {
+ return specId;
+ }
+
+ @Override
+ public NameIdentifier nameIdentifier() {
+ return name;
+ }
+
+ @Override
+ public String jobTemplateName() {
+ return IcebergRewriteManifestsContent.JOB_TEMPLATE_NAME_VALUE;
+ }
+
+ @Override
+ public Map<String, String> jobOptions() {
+ // An empty option clears shared submitter settings and preserves
Iceberg's runtime default.
+ return Map.of(
+ "spec_id",
+ Integer.toString(specId),
+ "use_caching",
+ useCaching == null ? "" : useCaching.toString());
+ }
+}
diff --git
a/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/ManifestRewriteStrategyHandler.java
b/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/ManifestRewriteStrategyHandler.java
new file mode 100644
index 0000000000..06a2c03a85
--- /dev/null
+++
b/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/ManifestRewriteStrategyHandler.java
@@ -0,0 +1,212 @@
+/*
+ * 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;
+
+import static
org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants.CURRENT_SNAPSHOT_ID;
+import static
org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants.DEFAULT_SPEC_ID;
+
+import com.google.common.base.Preconditions;
+import java.math.BigInteger;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import javax.annotation.Nullable;
+import org.apache.gravitino.maintenance.optimizer.api.common.StatisticEntry;
+import
org.apache.gravitino.maintenance.optimizer.api.recommender.StrategyEvaluation;
+import
org.apache.gravitino.maintenance.optimizer.api.recommender.StrategyHandler;
+import
org.apache.gravitino.maintenance.optimizer.api.recommender.StrategyHandlerContext;
+import
org.apache.gravitino.maintenance.optimizer.common.IcebergManifestStatistics;
+import org.apache.gravitino.policy.IcebergRewriteManifestsContent;
+import org.apache.gravitino.policy.PolicyContents;
+import org.apache.gravitino.stats.StatisticValue;
+import org.apache.gravitino.stats.StatisticValues;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** Evaluates a complete manifest measurement for exactly one resolved
partition spec. */
+public final class ManifestRewriteStrategyHandler implements StrategyHandler {
+ private static final Logger LOG =
LoggerFactory.getLogger(ManifestRewriteStrategyHandler.class);
+
+ private StrategyEvaluation evaluation = StrategyEvaluation.NO_EXECUTION;
+
+ @Override
+ public Set<DataRequirement> dataRequirements() {
+ return Set.of(DataRequirement.TABLE_METADATA,
DataRequirement.TABLE_STATISTICS);
+ }
+
+ @Override
+ public String strategyType() {
+ return IcebergRewriteManifestsContent.STRATEGY_TYPE_VALUE;
+ }
+
+ @Override
+ public void initialize(StrategyHandlerContext context) {
+ evaluation = StrategyEvaluation.NO_EXECUTION;
+ Objects.requireNonNull(context, "context");
+ Preconditions.checkArgument(
+ strategyType().equals(context.strategy().strategyType()),
+ "Unexpected strategy type: %s",
+ context.strategy().strategyType());
+ Preconditions.checkArgument(
+ IcebergRewriteManifestsContent.JOB_TEMPLATE_NAME_VALUE.equals(
+ context.strategy().jobTemplateName()),
+ "Unexpected manifest rewrite job template");
+ Map<String, Object> rules = context.strategy().rules();
+ Integer requestedSpec = requestedSpec(rules);
+ Integer resolvedSpec = requestedSpec == null ? defaultSpec(context) :
requestedSpec;
+ if (resolvedSpec == null) {
+ LOG.warn(
+ "Skipping manifest rewrite policy {} for {}: cannot resolve Iceberg
{} from table metadata",
+ context.strategy().name(),
+ context.nameIdentifier(),
+ DEFAULT_SPEC_ID);
+ return;
+ }
+ IcebergRewriteManifestsContent content =
+ PolicyContents.icebergRewriteManifests(
+ longRule(rules,
IcebergRewriteManifestsContent.MANIFEST_COUNT_CRITICAL),
+ longRule(rules,
IcebergRewriteManifestsContent.MANIFEST_COUNT_WARNING),
+ longRule(rules,
IcebergRewriteManifestsContent.AVG_MANIFEST_SIZE_THRESHOLD_BYTES),
+ requestedSpec,
+ booleanRule(rules, IcebergRewriteManifestsContent.USE_CACHING));
+ content.validate();
+ Map<String, StatisticValue<?>> statistics = new LinkedHashMap<>();
+ for (StatisticEntry<?> entry : context.tableStatistics()) {
+ Preconditions.checkArgument(
+ statistics.putIfAbsent(entry.name(), entry.value()) == null,
+ "Duplicate statistic: %s",
+ entry.name());
+ }
+ if (!hasCurrentSnapshot(context,
statistics.get(IcebergManifestStatistics.SNAPSHOT_ID))) {
+ LOG.info(
+ "Skipping manifest rewrite for {}: verify table snapshot metadata
and recollect manifest statistics",
+ context.nameIdentifier());
+ return;
+ }
+ Optional<IcebergManifestStatistics> measurement =
+ IcebergManifestStatistics.fromStatistics(statistics, resolvedSpec);
+ if (measurement.isEmpty()) {
+ LOG.info(
+ "Collect manifest statistics for spec {} of {} before evaluating",
+ resolvedSpec,
+ context.nameIdentifier());
+ return;
+ }
+ IcebergManifestStatistics values = measurement.get();
+ if (values.count() >= content.manifestCountCritical()
+ || (values.count() >= content.manifestCountWarning()
+ && values.averageSize() <
content.avgManifestSizeThresholdBytes())) {
+ evaluation =
+ new StrategyEvaluationImpl(
+ values.count(),
+ new ManifestRewriteJobContext(
+ context.nameIdentifier(), resolvedSpec,
content.useCaching()));
+ }
+ }
+
+ @Override
+ public boolean shouldTrigger() {
+ return evaluation.jobExecutionContext().isPresent();
+ }
+
+ @Override
+ public StrategyEvaluation evaluate() {
+ return evaluation;
+ }
+
+ private static boolean hasCurrentSnapshot(
+ StrategyHandlerContext context, @Nullable StatisticValue<?> snapshot) {
+ if (!(snapshot instanceof StatisticValues.LongValue)) {
+ return false;
+ }
+ String currentSnapshot =
+ context
+ .tableMetadata()
+ .map(table -> table.properties().get(CURRENT_SNAPSHOT_ID))
+ .orElse(null);
+ long collectedSnapshot = ((StatisticValues.LongValue) snapshot).value();
+ return collectedSnapshot == -1L
+ ? "none".equals(currentSnapshot)
+ : collectedSnapshot >= 0 &&
Long.toString(collectedSnapshot).equals(currentSnapshot);
+ }
+
+ @Nullable
+ private static Integer defaultSpec(StrategyHandlerContext context) {
+ String value =
+ context.tableMetadata().map(table ->
table.properties().get(DEFAULT_SPEC_ID)).orElse(null);
+ if (value == null || !value.matches("[0-9]+")) {
+ return null;
+ }
+ try {
+ return Integer.valueOf(value);
+ } catch (NumberFormatException e) {
+ return null;
+ }
+ }
+
+ @Nullable
+ private static Integer requestedSpec(Map<String, Object> rules) {
+ Long value = longRule(rules, IcebergRewriteManifestsContent.SPEC_ID);
+ Preconditions.checkArgument(
+ value == null || (value >= 0 && value <= Integer.MAX_VALUE),
+ "spec_id must be a non-negative 32-bit integer");
+ return value == null ? null : value.intValue();
+ }
+
+ @Nullable
+ private static Long longRule(Map<String, Object> rules, String key) {
+ Object value = rules.get(key);
+ if (value == null) {
+ return null;
+ }
+ if (value instanceof Byte
+ || value instanceof Short
+ || value instanceof Integer
+ || value instanceof Long) {
+ return ((Number) value).longValue();
+ }
+ try {
+ if (value instanceof BigInteger) {
+ return ((BigInteger) value).longValueExact();
+ }
+ Preconditions.checkArgument(
+ value instanceof String && ((String) value).matches("-?[0-9]+"),
+ "%s must be an integer",
+ key);
+ return Long.valueOf((String) value);
+ } catch (NumberFormatException | ArithmeticException e) {
+ throw new IllegalArgumentException(key + " must be an integer", e);
+ }
+ }
+
+ @Nullable
+ private static Boolean booleanRule(Map<String, Object> rules, String key) {
+ Object value = rules.get(key);
+ if (value == null) {
+ return null;
+ }
+ Preconditions.checkArgument(
+ "true".equalsIgnoreCase(value.toString()) ||
"false".equalsIgnoreCase(value.toString()),
+ "%s must be true or false",
+ key);
+ return Boolean.valueOf(value.toString());
+ }
+}
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 7b1cceeaf5..69fb475611 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
@@ -32,6 +32,7 @@ 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;
+import org.apache.gravitino.policy.IcebergRewriteManifestsContent;
/** Submits optimizer jobs to Gravitino using job template adapters. */
public class GravitinoJobSubmitter implements JobSubmitter {
@@ -51,6 +52,8 @@ public class GravitinoJobSubmitter implements JobSubmitter {
ImmutableMap.of(
IcebergDataCompactionContent.JOB_TEMPLATE_NAME_VALUE,
GravitinoCompactionJobAdapter.class,
+ IcebergRewriteManifestsContent.JOB_TEMPLATE_NAME_VALUE,
+ GravitinoManifestRewriteJobAdapter.class,
IcebergOrphanFileRemovalContent.JOB_TEMPLATE_NAME_VALUE,
GravitinoOrphanFileRemovalJobAdapter.class);
diff --git
a/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/job/GravitinoManifestRewriteJobAdapter.java
b/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/job/GravitinoManifestRewriteJobAdapter.java
new file mode 100644
index 0000000000..7186d5fe34
--- /dev/null
+++
b/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/job/GravitinoManifestRewriteJobAdapter.java
@@ -0,0 +1,46 @@
+/*
+ * 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.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.ManifestRewriteJobContext;
+
+/** Maps a manifest rewrite decision to the built-in Spark job's arguments. */
+public final class GravitinoManifestRewriteJobAdapter implements
GravitinoJobAdapter {
+ @Override
+ public Map<String, String> jobConfig(JobExecutionContext context) {
+ Preconditions.checkArgument(
+ context instanceof ManifestRewriteJobContext,
+ "jobExecutionContext must be ManifestRewriteJobContext");
+ ManifestRewriteJobContext rewrite = (ManifestRewriteJobContext) context;
+ IdentifierUtils.requireTableIdentifierNormalized(rewrite.nameIdentifier());
+ Map<String, String> config = new LinkedHashMap<>(rewrite.jobOptions());
+ config.put(
+ "catalog_name",
+
IdentifierUtils.getCatalogNameFromTableIdentifier(rewrite.nameIdentifier()));
+ config.put(
+ "table_identifier",
+
IdentifierUtils.removeCatalogFromIdentifier(rewrite.nameIdentifier()).toString());
+ return Map.copyOf(config);
+ }
+}
diff --git
a/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/recommender/TestManifestRewriteRecommendation.java
b/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/recommender/TestManifestRewriteRecommendation.java
new file mode 100644
index 0000000000..1890daa451
--- /dev/null
+++
b/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/recommender/TestManifestRewriteRecommendation.java
@@ -0,0 +1,334 @@
+/*
+ * 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;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Stream;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.maintenance.optimizer.api.common.StatisticEntry;
+import
org.apache.gravitino.maintenance.optimizer.api.recommender.JobExecutionContext;
+import org.apache.gravitino.maintenance.optimizer.api.recommender.JobSubmitter;
+import
org.apache.gravitino.maintenance.optimizer.api.recommender.StrategyProvider;
+import
org.apache.gravitino.maintenance.optimizer.api.recommender.SupportTableStatistics;
+import
org.apache.gravitino.maintenance.optimizer.api.recommender.TableMetadataProvider;
+import
org.apache.gravitino.maintenance.optimizer.common.IcebergManifestStatistics;
+import org.apache.gravitino.maintenance.optimizer.common.OptimizerEnv;
+import org.apache.gravitino.maintenance.optimizer.common.StatisticEntryImpl;
+import org.apache.gravitino.maintenance.optimizer.common.conf.OptimizerConfig;
+import
org.apache.gravitino.maintenance.optimizer.recommender.job.GravitinoManifestRewriteJobAdapter;
+import
org.apache.gravitino.maintenance.optimizer.recommender.strategy.GravitinoStrategy;
+import org.apache.gravitino.policy.IcebergRewriteManifestsContent;
+import org.apache.gravitino.policy.Policy;
+import org.apache.gravitino.policy.PolicyContent;
+import org.apache.gravitino.policy.PolicyContents;
+import org.apache.gravitino.rel.Table;
+import org.apache.gravitino.stats.StatisticValue;
+import org.apache.gravitino.stats.StatisticValues;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+import org.junit.jupiter.params.provider.NullSource;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+
+class TestManifestRewriteRecommendation {
+ @Test
+ void testBuiltInDiscoveryDryRunAndSubmission() throws Exception {
+ NameIdentifier identifier = NameIdentifier.of("catalog", "db", "table");
+ Policy policy = Mockito.mock(Policy.class);
+ Mockito.when(policy.name()).thenReturn("rewrite");
+ Mockito.when(policy.content())
+ .thenReturn(PolicyContents.icebergRewriteManifests(null, null, null,
7, false));
+ GravitinoStrategy strategy = new GravitinoStrategy(policy);
+ StrategyProvider strategies = Mockito.mock(StrategyProvider.class);
+
Mockito.when(strategies.strategies(identifier)).thenReturn(List.of(strategy));
+ Mockito.when(strategies.strategy("rewrite")).thenReturn(strategy);
+ SupportTableStatistics statistics =
Mockito.mock(SupportTableStatistics.class);
+ Mockito.when(statistics.tableStatistics(identifier))
+ .thenReturn(
+ List.of(
+ new StatisticEntryImpl<>(
+ IcebergManifestStatistics.MANIFEST_NUMBER,
+ StatisticValues.objectValue(Map.of("7",
StatisticValues.longValue(500)))),
+ new StatisticEntryImpl<>(
+ IcebergManifestStatistics.AVG_MANIFEST_SIZE,
+ StatisticValues.objectValue(
+ Map.of("7", StatisticValues.doubleValue(16 * 1024 *
1024)))),
+ new StatisticEntryImpl<>(
+ IcebergManifestStatistics.SNAPSHOT_ID,
StatisticValues.longValue(1))));
+ TableMetadataProvider metadata = Mockito.mock(TableMetadataProvider.class);
+ Table table = Mockito.mock(Table.class);
+ Mockito.when(table.properties()).thenReturn(Map.of("current-snapshot-id",
"1"));
+ Mockito.when(metadata.tableMetadata(identifier)).thenReturn(table);
+ JobSubmitter submitter = Mockito.mock(JobSubmitter.class);
+ Mockito.when(submitter.submitJob(Mockito.anyString(),
Mockito.any())).thenReturn("job-1");
+ try (Recommender recommender =
+ new Recommender(
+ strategies,
+ statistics,
+ metadata,
+ submitter,
+ new OptimizerEnv(new OptimizerConfig(Map.of())))) {
+ Assertions.assertEquals(
+ 1, recommender.recommendForStrategyName(List.of(identifier),
"rewrite", 1).size());
+ Mockito.verifyNoInteractions(submitter);
+ PolicyContent validContent = policy.content();
+ for (Object invalid : new Object[] {"+500", new StringBuilder("500"),
-1L}) {
+ Mockito.when(policy.content())
+ .thenReturn(
+ PolicyContents.custom(
+ Map.of("spec_id", 7, "manifest_count_critical", invalid),
+ validContent.supportedObjectTypes(),
+ validContent.properties()));
+ Assertions.assertTrue(
+ recommender.recommendForStrategyName(List.of(identifier),
"rewrite", 1).isEmpty());
+ }
+ Mockito.when(policy.content()).thenReturn(validContent);
+ List<Recommender.RecommendationResult> results =
+ recommender.submitForStrategyName(List.of(identifier), "rewrite");
+ Assertions.assertEquals("job-1", results.get(0).jobId());
+ ArgumentCaptor<JobExecutionContext> captured =
+ ArgumentCaptor.forClass(JobExecutionContext.class);
+ Mockito.verify(submitter)
+ .submitJob(
+
Mockito.eq(IcebergRewriteManifestsContent.JOB_TEMPLATE_NAME_VALUE),
+ captured.capture());
+ Assertions.assertEquals(
+ Map.of(
+ "catalog_name",
+ "catalog",
+ "table_identifier",
+ "db.table",
+ "spec_id",
+ "7",
+ "use_caching",
+ "false"),
+ new
GravitinoManifestRewriteJobAdapter().jobConfig(captured.getValue()));
+ Mockito.verify(metadata,
Mockito.atLeastOnce()).tableMetadata(identifier);
+
Mockito.when(table.properties()).thenReturn(Map.of("current-snapshot-id", "2"));
+ Assertions.assertTrue(
+ recommender.submitForStrategyName(List.of(identifier),
"rewrite").isEmpty());
+ Mockito.verifyNoMoreInteractions(submitter);
+ Mockito.verify(statistics,
Mockito.never()).partitionStatistics(Mockito.any());
+
Mockito.when(statistics.tableStatistics(identifier)).thenReturn(List.of());
+ Assertions.assertTrue(
+ recommender.submitForStrategyName(List.of(identifier),
"rewrite").isEmpty());
+ Mockito.verifyNoMoreInteractions(submitter);
+
Mockito.when(policy.content()).thenReturn(PolicyContents.icebergRewriteManifests());
+ Assertions.assertTrue(
+ recommender.submitForStrategyName(List.of(identifier),
"rewrite").isEmpty());
+ Mockito.verifyNoMoreInteractions(submitter);
+ }
+ }
+
+ @ParameterizedTest
+ @MethodSource("invalidMeasurements")
+ void testInvalidStatisticsDoNotAbortOtherTables(List<StatisticEntry<?>>
invalid)
+ throws Exception {
+ NameIdentifier bad = NameIdentifier.of("catalog", "db", "bad");
+ NameIdentifier healthy = NameIdentifier.of("catalog", "db", "healthy");
+ StrategyProvider strategies = manifestStrategies(bad, healthy);
+ SupportTableStatistics statistics =
Mockito.mock(SupportTableStatistics.class);
+ Mockito.when(statistics.tableStatistics(bad)).thenReturn(invalid);
+ Mockito.when(statistics.tableStatistics(healthy))
+ .thenReturn(measurements(StatisticValues.longValue(500), 1));
+ TableMetadataProvider metadata = Mockito.mock(TableMetadataProvider.class);
+ Table table = tableAtSnapshot("1");
+ Mockito.when(metadata.tableMetadata(Mockito.any())).thenReturn(table);
+ JobSubmitter submitter = Mockito.mock(JobSubmitter.class);
+ Mockito.when(submitter.submitJob(Mockito.anyString(),
Mockito.any())).thenReturn("healthy-job");
+ try (Recommender recommender =
+ new Recommender(
+ strategies,
+ statistics,
+ metadata,
+ submitter,
+ new OptimizerEnv(new OptimizerConfig(Map.of())))) {
+ List<Recommender.RecommendationResult> results =
+ recommender.submitForStrategyName(List.of(bad, healthy), "rewrite");
+ Assertions.assertEquals(1, results.size());
+ Assertions.assertEquals(healthy, results.get(0).identifier());
+ Assertions.assertEquals("healthy-job", results.get(0).jobId());
+ ArgumentCaptor<JobExecutionContext> captured =
+ ArgumentCaptor.forClass(JobExecutionContext.class);
+ Mockito.verify(submitter)
+ .submitJob(
+
Mockito.eq(IcebergRewriteManifestsContent.JOB_TEMPLATE_NAME_VALUE),
+ captured.capture());
+ Assertions.assertEquals(healthy, captured.getValue().nameIdentifier());
+ }
+ }
+
+ @Test
+ void testSnapshotChangeDuringStatisticsReadSkipsRecommendation() throws
Exception {
+ NameIdentifier identifier = NameIdentifier.of("catalog", "db", "table");
+ AtomicReference<Table> currentTable = new
AtomicReference<>(tableAtSnapshot("1"));
+ SupportTableStatistics statistics =
Mockito.mock(SupportTableStatistics.class);
+ Mockito.when(statistics.tableStatistics(identifier))
+ .thenAnswer(
+ invocation -> {
+ currentTable.set(tableAtSnapshot("2"));
+ return measurements(StatisticValues.longValue(500), 1);
+ });
+ TableMetadataProvider metadata = Mockito.mock(TableMetadataProvider.class);
+ Mockito.when(metadata.tableMetadata(identifier)).thenAnswer(invocation ->
currentTable.get());
+ JobSubmitter submitter = Mockito.mock(JobSubmitter.class);
+ try (Recommender recommender =
+ new Recommender(
+ manifestStrategies(identifier),
+ statistics,
+ metadata,
+ submitter,
+ new OptimizerEnv(new OptimizerConfig(Map.of())))) {
+ Assertions.assertTrue(
+ recommender.submitForStrategyName(List.of(identifier),
"rewrite").isEmpty());
+ Mockito.verifyNoInteractions(submitter);
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"statistics", "metadata"})
+ void testProviderFailuresStillPropagate(String failingProvider) throws
Exception {
+ NameIdentifier identifier = NameIdentifier.of("catalog", "db", "table");
+ IllegalArgumentException failure = new IllegalArgumentException("Provider
read failed");
+ SupportTableStatistics statistics =
Mockito.mock(SupportTableStatistics.class);
+ Mockito.when(statistics.tableStatistics(identifier))
+ .thenReturn(measurements(StatisticValues.longValue(500), 1));
+ TableMetadataProvider metadata = Mockito.mock(TableMetadataProvider.class);
+ if ("statistics".equals(failingProvider)) {
+ Mockito.when(statistics.tableStatistics(identifier)).thenThrow(failure);
+ } else {
+ Mockito.when(metadata.tableMetadata(identifier)).thenThrow(failure);
+ }
+ JobSubmitter submitter = Mockito.mock(JobSubmitter.class);
+ try (Recommender recommender =
+ new Recommender(
+ manifestStrategies(identifier),
+ statistics,
+ metadata,
+ submitter,
+ new OptimizerEnv(new OptimizerConfig(Map.of())))) {
+ Assertions.assertSame(
+ failure,
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> recommender.submitForStrategyName(List.of(identifier),
"rewrite")));
+ Mockito.verifyNoInteractions(submitter);
+ }
+ }
+
+ @ParameterizedTest
+ @NullSource
+ @ValueSource(strings = {"", "-1", "1.5", "+1", " 1", "2147483648",
"invalid"})
+ void testUnresolvableDefaultSkipsOnlyAffectedTable(String defaultSpec)
throws Exception {
+ NameIdentifier bad = NameIdentifier.of("catalog", "db", "bad");
+ NameIdentifier healthy = NameIdentifier.of("catalog", "db", "healthy");
+ SupportTableStatistics statistics =
Mockito.mock(SupportTableStatistics.class);
+ Mockito.when(statistics.tableStatistics(Mockito.any()))
+ .thenReturn(measurements(StatisticValues.longValue(500), 1));
+ TableMetadataProvider metadata = Mockito.mock(TableMetadataProvider.class);
+ Table badTable = Mockito.mock(Table.class);
+ Map<String, String> properties = new HashMap<>();
+ properties.put("current-snapshot-id", "1");
+ if (defaultSpec != null) {
+ properties.put("default-spec-id", defaultSpec);
+ }
+ Mockito.when(badTable.properties()).thenReturn(properties);
+ Table healthyTable = tableAtSnapshot("1");
+ Mockito.when(metadata.tableMetadata(bad)).thenReturn(badTable);
+ Mockito.when(metadata.tableMetadata(healthy)).thenReturn(healthyTable);
+ JobSubmitter submitter = Mockito.mock(JobSubmitter.class);
+ try (Recommender recommender =
+ new Recommender(
+ manifestStrategies(PolicyContents.icebergRewriteManifests(), bad,
healthy),
+ statistics,
+ metadata,
+ submitter,
+ new OptimizerEnv(new OptimizerConfig(Map.of())))) {
+ List<Recommender.RecommendationResult> results =
+ recommender.submitForStrategyName(List.of(bad, healthy), "rewrite");
+ Assertions.assertEquals(1, results.size());
+ Assertions.assertEquals(healthy, results.get(0).identifier());
+ Assertions.assertEquals("7", results.get(0).jobOptions().get("spec_id"));
+ ArgumentCaptor<JobExecutionContext> captured =
+ ArgumentCaptor.forClass(JobExecutionContext.class);
+ Mockito.verify(submitter)
+ .submitJob(
+
Mockito.eq(IcebergRewriteManifestsContent.JOB_TEMPLATE_NAME_VALUE),
+ captured.capture());
+ Assertions.assertEquals("7",
captured.getValue().jobOptions().get("spec_id"));
+ }
+ }
+
+ private static Stream<List<StatisticEntry<?>>> invalidMeasurements() {
+ return Stream.of(
+ measurements(StatisticValues.stringValue("500"), 1),
+ measurements(StatisticValues.longValue(-1), 1),
+ measurements(StatisticValues.longValue(500), Double.NaN),
+ measurements(StatisticValues.longValue(0), 1),
+ List.of(
+ new StatisticEntryImpl<>(
+ IcebergManifestStatistics.MANIFEST_NUMBER,
StatisticValues.longValue(500)),
+ new StatisticEntryImpl<>(
+ IcebergManifestStatistics.SNAPSHOT_ID,
StatisticValues.longValue(1))));
+ }
+
+ private static List<StatisticEntry<?>> measurements(StatisticValue<?> count,
double average) {
+ return List.of(
+ new StatisticEntryImpl<>(
+ IcebergManifestStatistics.MANIFEST_NUMBER,
+ StatisticValues.objectValue(Map.of("7", count))),
+ new StatisticEntryImpl<>(
+ IcebergManifestStatistics.AVG_MANIFEST_SIZE,
+ StatisticValues.objectValue(Map.of("7",
StatisticValues.doubleValue(average)))),
+ new StatisticEntryImpl<>(
+ IcebergManifestStatistics.SNAPSHOT_ID,
StatisticValues.longValue(1)));
+ }
+
+ private static StrategyProvider manifestStrategies(NameIdentifier...
identifiers) {
+ return manifestStrategies(
+ PolicyContents.icebergRewriteManifests(null, null, null, 7, null),
identifiers);
+ }
+
+ private static StrategyProvider manifestStrategies(
+ PolicyContent content, NameIdentifier... identifiers) {
+ Policy policy = Mockito.mock(Policy.class);
+ Mockito.when(policy.name()).thenReturn("rewrite");
+ Mockito.when(policy.content()).thenReturn(content);
+ GravitinoStrategy strategy = new GravitinoStrategy(policy);
+ StrategyProvider strategies = Mockito.mock(StrategyProvider.class);
+ Mockito.when(strategies.strategy("rewrite")).thenReturn(strategy);
+ for (NameIdentifier identifier : identifiers) {
+
Mockito.when(strategies.strategies(identifier)).thenReturn(List.of(strategy));
+ }
+ return strategies;
+ }
+
+ private static Table tableAtSnapshot(String snapshot) {
+ Table table = Mockito.mock(Table.class);
+ Mockito.when(table.properties())
+ .thenReturn(Map.of("current-snapshot-id", snapshot, "default-spec-id",
"7"));
+ return table;
+ }
+}
diff --git
a/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/recommender/TestRecommenderExecutionMode.java
b/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/recommender/TestRecommenderExecutionMode.java
index 1edf9492ba..9cb224c97d 100644
---
a/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/recommender/TestRecommenderExecutionMode.java
+++
b/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/recommender/TestRecommenderExecutionMode.java
@@ -36,6 +36,8 @@ import
org.apache.gravitino.maintenance.optimizer.common.OptimizerEnv;
import org.apache.gravitino.maintenance.optimizer.common.conf.OptimizerConfig;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.Mockito;
class TestRecommenderExecutionMode {
@@ -210,6 +212,54 @@ class TestRecommenderExecutionMode {
Assertions.assertTrue(exception.getMessage().contains("limit must be >
0"));
}
+ @ParameterizedTest
+ @ValueSource(strings = {"initialize", "shouldTrigger", "evaluate"})
+ void testInvalidHandlerInputSkipsOnlyAffectedTable(String failureStage)
throws Exception {
+ NameIdentifier bad = NameIdentifier.of("catalog", "db", failureStage);
+ NameIdentifier healthy = NameIdentifier.of("catalog", "db", "healthy");
+ Strategy strategy = new TestStrategy("s1", STRATEGY_TYPE, "tpl-1");
+ StrategyProvider strategies = Mockito.mock(StrategyProvider.class);
+
Mockito.when(strategies.strategies(Mockito.any())).thenReturn(List.of(strategy));
+ Mockito.when(strategies.strategy("s1")).thenReturn(strategy);
+ JobSubmitter submitter = Mockito.mock(JobSubmitter.class);
+ try (Recommender recommender =
+ createRecommender(
+ strategies,
+ Mockito.mock(StatisticsProvider.class),
+ Mockito.mock(TableMetadataProvider.class),
+ submitter)) {
+ List<Recommender.RecommendationResult> results =
+ recommender.recommendForStrategyName(List.of(bad, healthy), "s1",
Integer.MAX_VALUE);
+ Assertions.assertEquals(1, results.size());
+ Assertions.assertEquals(healthy, results.get(0).identifier());
+ Mockito.verifyNoInteractions(submitter);
+ }
+ }
+
+ @Test
+ void testMissingHandlerConfigurationStillFails() throws Exception {
+ NameIdentifier identifier = NameIdentifier.of("catalog", "db", "table");
+ Strategy strategy = new TestStrategy("s1", STRATEGY_TYPE, "tpl-1");
+ StrategyProvider strategies = Mockito.mock(StrategyProvider.class);
+
Mockito.when(strategies.strategies(identifier)).thenReturn(List.of(strategy));
+ Mockito.when(strategies.strategy("s1")).thenReturn(strategy);
+ try (Recommender recommender =
+ new Recommender(
+ strategies,
+ Mockito.mock(StatisticsProvider.class),
+ Mockito.mock(TableMetadataProvider.class),
+ Mockito.mock(JobSubmitter.class),
+ new OptimizerEnv(new OptimizerConfig(Map.of())))) {
+ IllegalArgumentException failure =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ recommender.recommendForStrategyName(
+ List.of(identifier), "s1", Integer.MAX_VALUE));
+ Assertions.assertTrue(failure.getMessage().contains("No StrategyHandler
class configured"));
+ }
+ }
+
private static Recommender createRecommender(
StrategyProvider strategyProvider,
StatisticsProvider statisticsProvider,
@@ -224,12 +274,14 @@ class TestRecommenderExecutionMode {
strategyProvider, statisticsProvider, tableMetadataProvider,
jobSubmitter, optimizerEnv);
}
+ /** Handler that rejects the table named for a lifecycle stage at that
stage. */
public static class TestStrategyHandler implements StrategyHandler {
private StrategyHandlerContext context;
@Override
public void initialize(StrategyHandlerContext context) {
this.context = context;
+ rejectInvalidTable("initialize");
}
@Override
@@ -239,11 +291,13 @@ class TestRecommenderExecutionMode {
@Override
public boolean shouldTrigger() {
+ rejectInvalidTable("shouldTrigger");
return true;
}
@Override
public StrategyEvaluation evaluate() {
+ rejectInvalidTable("evaluate");
Strategy strategy = context.strategy();
JobExecutionContext jobExecutionContext =
new JobExecutionContext() {
@@ -274,6 +328,12 @@ class TestRecommenderExecutionMode {
}
};
}
+
+ private void rejectInvalidTable(String stage) {
+ if (stage.equals(context.nameIdentifier().name())) {
+ throw new IllegalArgumentException("Invalid table input at " + stage);
+ }
+ }
}
private static final class TestStrategy implements Strategy {
diff --git
a/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/TestManifestRewriteStrategyHandler.java
b/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/TestManifestRewriteStrategyHandler.java
new file mode 100644
index 0000000000..a8fd356aab
--- /dev/null
+++
b/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/TestManifestRewriteStrategyHandler.java
@@ -0,0 +1,384 @@
+/*
+ * 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;
+
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.util.List;
+import java.util.Map;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.maintenance.optimizer.api.common.StatisticEntry;
+import
org.apache.gravitino.maintenance.optimizer.api.recommender.StrategyHandlerContext;
+import
org.apache.gravitino.maintenance.optimizer.common.IcebergManifestStatistics;
+import org.apache.gravitino.maintenance.optimizer.common.StatisticEntryImpl;
+import
org.apache.gravitino.maintenance.optimizer.recommender.job.GravitinoManifestRewriteJobAdapter;
+import
org.apache.gravitino.maintenance.optimizer.recommender.strategy.GravitinoStrategy;
+import org.apache.gravitino.policy.Policy;
+import org.apache.gravitino.policy.PolicyContent;
+import org.apache.gravitino.policy.PolicyContents;
+import org.apache.gravitino.rel.Table;
+import org.apache.gravitino.stats.StatisticValues;
+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;
+import org.mockito.Mockito;
+
+class TestManifestRewriteStrategyHandler {
+ @ParameterizedTest
+ @CsvSource({
+ "0,0,false",
+ "99,1,false",
+ "100,8388607,true",
+ "100,8388608,false",
+ "100,8388609,false",
+ "499,8388607,true",
+ "499,8388608,false",
+ "500,8388608,true",
+ "500,16777216,true",
+ "501,16777216,true"
+ })
+ void testBoundaries(long count, double average, boolean trigger) {
+ ManifestRewriteStrategyHandler handler = new
ManifestRewriteStrategyHandler();
+ handler.initialize(
+ context(
+ PolicyContents.icebergRewriteManifests(null, null, null, 1, null),
+ statistics(count, average)));
+ Assertions.assertEquals(trigger, handler.shouldTrigger());
+ Assertions.assertEquals(trigger,
handler.evaluate().jobExecutionContext().isPresent());
+ if (trigger) {
+ Assertions.assertEquals(count, handler.evaluate().score());
+ }
+ }
+
+ @Test
+ void testMissingSpecAndPartialMeasurementsDoNotTrigger() {
+ ManifestRewriteStrategyHandler handler = new
ManifestRewriteStrategyHandler();
+ for (List<StatisticEntry<?>> stats :
+ List.of(
+ List.<StatisticEntry<?>>of(),
+ List.of(statistics(500, 1).get(0), statistics(500, 1).get(2)),
+ statistics(500, 1).subList(1, 3))) {
+ handler.initialize(
+ context(PolicyContents.icebergRewriteManifests(null, null, null, 1,
null), stats));
+ Assertions.assertFalse(handler.shouldTrigger());
+ }
+ handler.initialize(
+ context(
+ PolicyContents.icebergRewriteManifests(null, null, null, 2, null),
statistics(500, 1)));
+ Assertions.assertFalse(handler.shouldTrigger());
+ }
+
+ @Test
+ void testMissingDefaultSkipsButExplicitSpecRetainsOptions() {
+ ManifestRewriteStrategyHandler handler = new
ManifestRewriteStrategyHandler();
+ StrategyHandlerContext context =
+ context(
+ PolicyContents.icebergRewriteManifests(null, null, null, null,
false),
+ statistics(100, 1));
+ handler.initialize(context);
+ Assertions.assertFalse(handler.shouldTrigger());
+ handler.initialize(
+ context(
+ PolicyContents.icebergRewriteManifests(null, null, null, 1, false),
+ statistics(100, 1)));
+ Map<String, String> config =
+ new GravitinoManifestRewriteJobAdapter()
+ .jobConfig(handler.evaluate().jobExecutionContext().orElseThrow());
+ Assertions.assertEquals(
+ Map.of(
+ "catalog_name",
+ "iceberg",
+ "table_identifier",
+ "db.tbl",
+ "spec_id",
+ "1",
+ "use_caching",
+ "false"),
+ config);
+ handler.initialize(context);
+ Assertions.assertFalse(handler.shouldTrigger());
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"0", "1", "2147483647"})
+ void testDefaultSpecIsResolvedAndPinned(String specId) {
+ Table metadata = Mockito.mock(Table.class);
+ Mockito.when(metadata.properties())
+ .thenReturn(Map.of("current-snapshot-id", "1", "default-spec-id",
specId));
+ StrategyHandlerContext context =
context(PolicyContents.icebergRewriteManifests(), List.of());
+ ManifestRewriteStrategyHandler handler = new
ManifestRewriteStrategyHandler();
+ handler.initialize(
+ StrategyHandlerContext.builder(context.nameIdentifier(),
context.strategy())
+ .withTableMetadata(metadata)
+ .withTableStatistics(
+ List.of(
+ new StatisticEntryImpl<>(
+ IcebergManifestStatistics.MANIFEST_NUMBER,
+ StatisticValues.objectValue(
+ Map.of(specId, StatisticValues.longValue(500)))),
+ new StatisticEntryImpl<>(
+ IcebergManifestStatistics.AVG_MANIFEST_SIZE,
+ StatisticValues.objectValue(
+ Map.of(specId, StatisticValues.doubleValue(1)))),
+ new StatisticEntryImpl<>(
+ IcebergManifestStatistics.SNAPSHOT_ID,
StatisticValues.longValue(1))))
+ .build());
+ Assertions.assertTrue(handler.shouldTrigger());
+ Mockito.when(metadata.properties())
+ .thenReturn(Map.of("current-snapshot-id", "1", "default-spec-id",
"2"));
+ Assertions.assertEquals(
+ specId,
+ new GravitinoManifestRewriteJobAdapter()
+ .jobConfig(handler.evaluate().jobExecutionContext().orElseThrow())
+ .get("spec_id"));
+ }
+
+ @Test
+ void testExplicitSpecOverridesCurrentDefault() {
+ StrategyHandlerContext context =
+ context(
+ PolicyContents.icebergRewriteManifests(null, null, null, 0, null),
statistics(0, 0));
+ Table metadata = Mockito.mock(Table.class);
+ Mockito.when(metadata.properties())
+ .thenReturn(Map.of("current-snapshot-id", "1", "default-spec-id",
"1"));
+ ManifestRewriteStrategyHandler handler = new
ManifestRewriteStrategyHandler();
+ handler.initialize(
+ StrategyHandlerContext.builder(context.nameIdentifier(),
context.strategy())
+ .withTableMetadata(metadata)
+ .withTableStatistics(context.tableStatistics())
+ .build());
+ Assertions.assertTrue(handler.shouldTrigger());
+ Assertions.assertEquals(
+ "0",
handler.evaluate().jobExecutionContext().orElseThrow().jobOptions().get("spec_id"));
+ }
+
+ @Test
+ void testConfigurableThresholds() {
+ ManifestRewriteStrategyHandler handler = new
ManifestRewriteStrategyHandler();
+ handler.initialize(
+ context(PolicyContents.icebergRewriteManifests(10L, 5L, 2L, 1, true),
statistics(5, 1)));
+ Assertions.assertTrue(handler.shouldTrigger());
+ }
+
+ @Test
+ void testOnlySelectedSpecCanTrigger() {
+ List<StatisticEntry<?>> measurements = statistics(0, 0);
+ ManifestRewriteStrategyHandler handler = new
ManifestRewriteStrategyHandler();
+ handler.initialize(
+ context(PolicyContents.icebergRewriteManifests(null, null, null, 1,
null), measurements));
+ Assertions.assertFalse(handler.shouldTrigger(), "Spec 1 has zero
manifests");
+ handler.initialize(
+ context(PolicyContents.icebergRewriteManifests(null, null, null, 0,
null), measurements));
+ Assertions.assertTrue(handler.shouldTrigger(), "Spec 0 has 1000
manifests");
+ Assertions.assertEquals(1000, handler.evaluate().score());
+ Assertions.assertEquals(
+ "0",
+ new GravitinoManifestRewriteJobAdapter()
+ .jobConfig(handler.evaluate().jobExecutionContext().orElseThrow())
+ .get("spec_id"));
+ }
+
+ @ParameterizedTest
+ @CsvSource({"true,true", "TRUE,true", "TrUe,true", "false,false",
"FALSE,false", "FaLsE,false"})
+ void testMixedCaseCachingReachesJobConfiguration(String input, String
expected) {
+ ManifestRewriteStrategyHandler handler = new
ManifestRewriteStrategyHandler();
+ handler.initialize(
+ context(customContent(Map.of("spec_id", 1, "use_caching", input)),
statistics(500, 1)));
+ Assertions.assertEquals(
+ expected,
+ new GravitinoManifestRewriteJobAdapter()
+ .jobConfig(handler.evaluate().jobExecutionContext().orElseThrow())
+ .get("use_caching"));
+ }
+
+ @Test
+ void testIntegralRuleValues() {
+ ManifestRewriteStrategyHandler handler = new
ManifestRewriteStrategyHandler();
+ for (Object warning :
+ new Object[] {(byte) 100, (short) 100, 100, 100L,
BigInteger.valueOf(100), "100"}) {
+ handler.initialize(
+ context(
+ customContent(Map.of("spec_id", 1, "manifest_count_warning",
warning)),
+ statistics(100, 1)));
+ Assertions.assertTrue(handler.shouldTrigger());
+ }
+ handler.initialize(
+ context(customContent(Map.of("spec_id", (long) Integer.MAX_VALUE)),
statistics(500, 1)));
+ Assertions.assertFalse(handler.shouldTrigger());
+ }
+
+ @Test
+ void testNonIntegerRuleValuesAreRejected() {
+ ManifestRewriteStrategyHandler handler = new
ManifestRewriteStrategyHandler();
+ for (Object warning :
+ new Object[] {
+ "+100",
+ "100.0",
+ "1e2",
+ " 100",
+ 100.0d,
+ 100.0f,
+ new BigDecimal("100"),
+ new StringBuilder("100"),
+ new BigInteger("9223372036854775808"),
+ "9223372036854775808"
+ }) {
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ handler.initialize(
+ context(
+ customContent(Map.of("spec_id", 1,
"manifest_count_warning", warning)),
+ statistics(100, 1))),
+ warning.toString());
+ }
+ }
+
+ @Test
+ void testMalformedRulesAndMeasurements() {
+ ManifestRewriteStrategyHandler handler = new
ManifestRewriteStrategyHandler();
+ for (Map<String, Object> rules :
+ List.<Map<String, Object>>of(
+ Map.of("spec_id", "2147483648"),
+ Map.of("spec_id", 2147483648L),
+ Map.of("spec_id", -1),
+ Map.of("spec_id", 1.5),
+ Map.of("spec_id", 1, "use_caching", "yes"),
+ Map.of("spec_id", 1, "manifest_count_critical", 99))) {
+ PolicyContent content = customContent(rules);
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> handler.initialize(context(content, statistics(500, 1))));
+ }
+ for (double average : new double[] {-1, Double.NaN,
Double.POSITIVE_INFINITY}) {
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ handler.initialize(
+ context(
+ PolicyContents.icebergRewriteManifests(null, null, null,
1, null),
+ statistics(500, average))));
+ }
+ }
+
+ @Test
+ void testUnverifiedSnapshotsDoNotTriggerAndClearEarlierEvaluation() {
+ ManifestRewriteStrategyHandler handler = new
ManifestRewriteStrategyHandler();
+ StrategyHandlerContext valid =
+ context(
+ PolicyContents.icebergRewriteManifests(null, null, null, 1, null),
statistics(500, 1));
+ for (Map<String, String> properties :
+ List.of(
+ Map.<String, String>of(), Map.of("current-snapshot-id", "2"),
+ Map.of("current-snapshot-id", "none"),
Map.of("current-snapshot-id", "invalid"))) {
+ handler.initialize(valid);
+ Assertions.assertTrue(handler.shouldTrigger());
+ Table metadata = Mockito.mock(Table.class);
+ Mockito.when(metadata.properties()).thenReturn(properties);
+ handler.initialize(
+ StrategyHandlerContext.builder(valid.nameIdentifier(),
valid.strategy())
+ .withTableStatistics(valid.tableStatistics())
+ .withTableMetadata(metadata)
+ .build());
+ Assertions.assertFalse(handler.shouldTrigger());
+ }
+ handler.initialize(
+ StrategyHandlerContext.builder(valid.nameIdentifier(),
valid.strategy())
+ .withTableStatistics(valid.tableStatistics())
+ .build());
+ Assertions.assertFalse(handler.shouldTrigger());
+ handler.initialize(
+ context(
+ PolicyContents.icebergRewriteManifests(null, null, null, 1, null),
+ statistics(500, 1).subList(0, 2)));
+ Assertions.assertFalse(handler.shouldTrigger());
+ handler.initialize(
+ context(
+ PolicyContents.icebergRewriteManifests(null, null, null, 1, null),
+ List.of(
+ statistics(500, 1).get(0),
+ statistics(500, 1).get(1),
+ new StatisticEntryImpl<>(
+ IcebergManifestStatistics.SNAPSHOT_ID,
StatisticValues.stringValue("1")))));
+ Assertions.assertFalse(handler.shouldTrigger());
+ }
+
+ @Test
+ void testEmptyTableSnapshotIsAccepted() {
+ StrategyHandlerContext valid =
+ context(
+ PolicyContents.icebergRewriteManifests(null, null, null, 1, null),
statistics(0, 0));
+ Table metadata = Mockito.mock(Table.class);
+
Mockito.when(metadata.properties()).thenReturn(Map.of("current-snapshot-id",
"none"));
+ ManifestRewriteStrategyHandler handler = new
ManifestRewriteStrategyHandler();
+ handler.initialize(
+ StrategyHandlerContext.builder(valid.nameIdentifier(),
valid.strategy())
+ .withTableMetadata(metadata)
+ .withTableStatistics(
+ List.of(
+ statistics(0, 0).get(0),
+ statistics(0, 0).get(1),
+ new StatisticEntryImpl<>(
+ IcebergManifestStatistics.SNAPSHOT_ID,
StatisticValues.longValue(-1))))
+ .build());
+ Assertions.assertFalse(handler.shouldTrigger());
+ }
+
+ private static PolicyContent customContent(Map<String, Object> rules) {
+ return PolicyContents.custom(
+ rules,
+ PolicyContents.icebergRewriteManifests().supportedObjectTypes(),
+ PolicyContents.icebergRewriteManifests().properties());
+ }
+
+ private static StrategyHandlerContext context(
+ PolicyContent content, List<StatisticEntry<?>> statistics) {
+ Policy policy = Mockito.mock(Policy.class);
+ Mockito.when(policy.content()).thenReturn(content);
+ Mockito.when(policy.name()).thenReturn("rewrite");
+ Table metadata = Mockito.mock(Table.class);
+
Mockito.when(metadata.properties()).thenReturn(Map.of("current-snapshot-id",
"1"));
+ return StrategyHandlerContext.builder(
+ NameIdentifier.of("iceberg", "db", "tbl"), new
GravitinoStrategy(policy))
+ .withTableMetadata(metadata)
+ .withTableStatistics(statistics)
+ .build();
+ }
+
+ private static List<StatisticEntry<?>> statistics(long count, double
average) {
+ return List.of(
+ new StatisticEntryImpl<>(
+ IcebergManifestStatistics.MANIFEST_NUMBER,
+ StatisticValues.objectValue(
+ Map.of(
+ "0", StatisticValues.longValue(1000), "1",
StatisticValues.longValue(count)))),
+ new StatisticEntryImpl<>(
+ IcebergManifestStatistics.AVG_MANIFEST_SIZE,
+ StatisticValues.objectValue(
+ Map.of(
+ "0",
+ StatisticValues.doubleValue(1),
+ "1",
+ StatisticValues.doubleValue(average)))),
+ new StatisticEntryImpl<>(
+ IcebergManifestStatistics.SNAPSHOT_ID,
StatisticValues.longValue(1)));
+ }
+}
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 2963fd003c..cebf7edbbf 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
@@ -24,12 +24,46 @@ import org.apache.gravitino.NameIdentifier;
import
org.apache.gravitino.maintenance.optimizer.api.recommender.JobExecutionContext;
import org.apache.gravitino.maintenance.optimizer.common.OptimizerEnv;
import org.apache.gravitino.maintenance.optimizer.common.conf.OptimizerConfig;
+import
org.apache.gravitino.maintenance.optimizer.recommender.handler.ManifestRewriteJobContext;
import org.apache.gravitino.policy.IcebergDataCompactionContent;
import org.apache.gravitino.policy.IcebergOrphanFileRemovalContent;
+import org.apache.gravitino.policy.IcebergRewriteManifestsContent;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.NullSource;
+import org.junit.jupiter.params.provider.ValueSource;
public class TestGravitinoJobSubmitter {
+ @ParameterizedTest
+ @NullSource
+ @ValueSource(booleans = {true, false})
+ void testManifestAdapterPreservesPolicyOptionsOverGlobalConfig(Boolean
useCaching) {
+ GravitinoJobAdapter adapter =
+ new GravitinoJobSubmitter()
+
.loadJobAdapter(IcebergRewriteManifestsContent.JOB_TEMPLATE_NAME_VALUE);
+ Assertions.assertInstanceOf(GravitinoManifestRewriteJobAdapter.class,
adapter);
+ OptimizerConfig config =
+ new OptimizerConfig(
+ Map.of(
+ OptimizerConfig.JOB_SUBMITTER_CONFIG_PREFIX + "spec_id", "99",
+ OptimizerConfig.JOB_SUBMITTER_CONFIG_PREFIX + "use_caching",
"false",
+ OptimizerConfig.JOB_SUBMITTER_CONFIG_PREFIX + "catalog_name",
"other",
+ OptimizerConfig.JOB_SUBMITTER_CONFIG_PREFIX + "spark_master",
"local[2]"));
+ Map<String, String> result =
+ GravitinoJobSubmitter.buildJobConfig(
+ config,
+ new ManifestRewriteJobContext(
+ NameIdentifier.of("catalog", "db", "table"), 1, useCaching),
+ adapter);
+ Assertions.assertEquals("1", result.get("spec_id"));
+ Assertions.assertEquals("catalog", result.get("catalog_name"));
+ Assertions.assertEquals("local[2]", result.get("spark_master"));
+ Assertions.assertEquals(
+ useCaching == null ? "" : useCaching.toString(),
result.get("use_caching"));
+ Assertions.assertThrows(IllegalArgumentException.class, () ->
adapter.jobConfig(null));
+ }
+
@Test
void loadJobAdapterReturnsCompactionAdapterForBuiltInTemplateName() {
GravitinoJobSubmitter submitter = new GravitinoJobSubmitter();
diff --git
a/server/src/test/java/org/apache/gravitino/server/web/rest/TestPolicyOperations.java
b/server/src/test/java/org/apache/gravitino/server/web/rest/TestPolicyOperations.java
index 0063630f7b..fbff3efd7f 100644
---
a/server/src/test/java/org/apache/gravitino/server/web/rest/TestPolicyOperations.java
+++
b/server/src/test/java/org/apache/gravitino/server/web/rest/TestPolicyOperations.java
@@ -81,6 +81,8 @@ import org.glassfish.jersey.test.TestProperties;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import org.mockito.Mockito;
public class TestPolicyOperations extends BaseOperationsTest {
@@ -409,6 +411,44 @@ public class TestPolicyOperations extends
BaseOperationsTest {
Assertions.assertEquals(RuntimeException.class.getSimpleName(),
errorResp1.getType());
}
+ @ParameterizedTest
+ @CsvSource({
+ "spec_id,1.5",
+ "manifest_count_critical,500.5",
+ "manifest_count_warning,100.5",
+ "avg_manifest_size_threshold_bytes,4096.5",
+ "use_caching,0",
+ "use_caching,1"
+ })
+ public void testCreateManifestPolicyRejectsInvalidFieldTypes(String field,
String value) {
+ PolicyEntity policy =
+ PolicyEntity.builder()
+ .withId(1L)
+ .withName("rewrite")
+ .withPolicyType(Policy.BuiltInType.ICEBERG_REWRITE_MANIFESTS)
+ .withEnabled(true)
+ .withContent(PolicyContents.icebergRewriteManifests())
+ .withAuditInfo(testAuditInfo1)
+ .build();
+ when(policyManager.createPolicy(any(), any(), any(), any(), anyBoolean(),
any()))
+ .thenReturn(policy);
+ String json =
+
"{\"name\":\"rewrite\",\"policyType\":\"system_iceberg_rewrite_manifests\","
+ + "\"enabled\":true,\"content\":{\""
+ + field
+ + "\":"
+ + value
+ + "}}";
+ try (Response response =
+ target(policyPath(metalake))
+ .request(MediaType.APPLICATION_JSON_TYPE)
+ .accept("application/vnd.gravitino.v1+json")
+ .post(Entity.entity(json, MediaType.APPLICATION_JSON_TYPE))) {
+ Assertions.assertEquals(Response.Status.BAD_REQUEST.getStatusCode(),
response.getStatus());
+ }
+ Mockito.verifyNoInteractions(policyManager);
+ }
+
@Test
public void testCreatePolicyWithIcebergCompactionType() {
PolicyContent content =
