laserninja commented on code in PR #13533:
URL: https://github.com/apache/gravitino/pull/13533#discussion_r4365990660
##########
common/src/main/java/org/apache/gravitino/dto/policy/PolicyContentDTO.java:
##########
@@ -307,19 +315,95 @@ private PolicyContent toDomainContent() {
}
}
- /** 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")
+ private Boolean useCaching;
Review Comment:
Fixed in 8e5163930. Added a field-specific deserializer that accepts JSON
booleans and preserves null/omitted defaults, while rejecting numbers, strings,
arrays, and objects. Create/update DTO tests cover these cases, and REST tests
now verify that numeric 0/1 return HTTP 400 without calling the policy manager.
##########
maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/ManifestRewriteJobContext.java:
##########
@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.maintenance.optimizer.recommender.handler;
+
+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() {
+ return useCaching == null
+ ? Map.of("spec_id", Integer.toString(specId))
+ : Map.of("spec_id", Integer.toString(specId), "use_caching",
useCaching.toString());
Review Comment:
Fixed in 8e5163930. A null policy caching setting now emits use_caching="",
overriding shared submitter configuration and letting the existing job parser
preserve Iceberg's runtime default. The submitter merge test covers null, true,
and false against a shared use_caching=false setting.
##########
maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/ManifestRewriteStrategyHandler.java:
##########
@@ -0,0 +1,215 @@
+/*
+ * 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 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;
+ Integer requestedSpec = requestedSpec(context.strategy().rules());
+ if (requestedSpec == null) {
+ LOG.warn(
+ "Skipping manifest rewrite policy {} for {}: set spec_id in the
policy before running the recommender",
+ context.strategy().name(),
+ context.nameIdentifier());
+ return;
+ }
+ initialize(context, requestedSpec);
+ }
+
+ /**
+ * Evaluates statistics using the spec resolved at collection start. The
default is never resolved
+ * again, even if partition evolution has occurred since collection.
+ *
+ * <p>This overload is for direct programmatic callers. The recommender and
CLI call {@link
+ * #initialize(StrategyHandlerContext)} and require an explicit policy spec
ID.
+ *
+ * @param context strategy, fresh table metadata, and one consistent
table-statistics response
+ * @param resolvedSpecId spec selected before collection
+ */
+ public void initialize(StrategyHandlerContext context, int resolvedSpecId) {
+ evaluation = StrategyEvaluation.NO_EXECUTION;
+ Objects.requireNonNull(context, "context");
+ Preconditions.checkArgument(resolvedSpecId >= 0, "spec_id must be >= 0");
+ 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);
+ Preconditions.checkArgument(
+ requestedSpec == null || requestedSpec == resolvedSpecId,
+ "Resolved spec ID does not match the policy spec_id");
+ 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();
Review Comment:
Fixed in 8e5163930. Recommender now logs and skips per-table
IllegalArgumentException failures from initialize, shouldTrigger, or evaluate,
then continues the batch. Provider reads and handler configuration remain
outside that catch. Tests cover five malformed-measurement cases followed by a
healthy table that still submits, invalid custom rules, all three handler
lifecycle stages, and propagation of provider/configuration errors.
##########
api/src/main/java/org/apache/gravitino/policy/IcebergRewriteManifestsContent.java:
##########
@@ -41,21 +39,108 @@ 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 requested partition spec ID, or null to use the cycle
target. */
Review Comment:
Fixed in 8e5163930. Updated SPEC_ID, specId(), and the default constructor
Javadocs: null stores defaults without a target, and the recommender skips the
policy until an explicit spec_id is set.
##########
api/src/main/java/org/apache/gravitino/policy/PolicyContents.java:
##########
@@ -171,12 +171,36 @@ public static PolicyContent icebergOrphanFileRemoval(
}
/**
- * Creates placeholder Iceberg rewrite-manifests policy content.
+ * Creates a manifest rewrite policy with default thresholds and cycle
target selection.
Review Comment:
Fixed in 8e5163930. Both factory Javadocs now describe storing default
thresholds without a target and explicitly state that the recommender skips
policies without spec_id. Removed the automatic cycle-target wording.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
