lasdf1234 commented on code in PR #13533:
URL: https://github.com/apache/gravitino/pull/13533#discussion_r4365911647
##########
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:
**P1: `use_caching` silently coerces numeric `0`/`1`.**
Thresholds and `spec_id` use strict deserializers, but `use_caching` relies
on Jackson’s default `Boolean` handling. In Jackson 2.13 (this repo),
`{"use_caching": 1}` becomes `true` and `0` becomes `false`, which undercuts
the “no silent coercion” story and OpenAPI `type: boolean`.
Please add a deserializer that accepts only JSON booleans (and optionally
the strings `"true"`/`"false"`), reject numbers, and cover this in
`TestManifestRewritePolicyDTO` / REST tests.
##########
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:
**P1: Omitted `use_caching` can leak from submitter configs.**
`GravitinoJobSubmitter.buildJobConfig` merges submitter configs first, then
adapter options. When policy `use_caching` is null, this method omits the key
entirely, so a global `jobSubmitterConfigs.use_caching` survives and is
forwarded to Spark — overriding the intended “leave Iceberg’s default / omit
the flag” behavior.
You already force-override `spec_id` / `catalog_name` for the same reason
(see `TestGravitinoJobSubmitter`). For null caching, emit `use_caching` → `""`
(or otherwise clear the key) so the template / `validateUseCaching` treat it as
absent and override submitter bleed-through. Please add a merge test with
submitter `use_caching=false` and policy null.
##########
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:
**P2: Same “cycle target selection” wording here.**
`Creates a manifest rewrite policy with default thresholds and cycle target
selection` / `null for the resolved collection-cycle target` implies automatic
target resolution that this PR does not implement on the recommend path. Please
reword to match `docs/iceberg-rewrite-manifests-policy.md` (recommender
requires an explicit `spec_id`; omitted policies are skipped).
##########
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:
**P2: Javadoc still describes a “cycle target” the recommender never uses.**
Product docs correctly say omitted `spec_id` is skipped by the
recommender/CLI. This constant (and `specId()`) still promise null → “use the
cycle target”, but production only calls the single-arg `initialize`, which
skips when `spec_id` is null. The two-arg overload that accepts a resolved spec
is for programmatic/IT callers only.
Please align the public API javadoc with the docs (null = stored defaults /
skip until explicit `spec_id`), or wire real cycle-target resolution — don’t
leave both stories.
##########
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:
**P1: One bad table can abort the whole recommend batch.**
Missing/stale snapshot and incomplete stats correctly fail open (skip +
log). But `content.validate()`, rule parsing, or
`IcebergManifestStatistics.fromStatistics(...)` throwing
`IllegalArgumentException` (NaN / wrong types / bad custom rules) propagates
out of `initialize`. `Recommender.recommendForOneStrategy` has no
per-identifier try/catch, so one corrupt table prevents recommendations for
every other identifier in the batch.
Prefer catching around initialize/shouldTrigger per identifier, log, skip
that table, and continue — same posture as snapshot mismatch. A two-identifier
test (first bad stats, second still recommends) would lock this in.
##########
maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/Recommender.java:
##########
@@ -371,7 +373,12 @@ private StrategyHandler createStrategyHandler(String
strategyType) {
* {@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();
Review Comment:
**P2 (nit): Built-in handler fallback looks good.**
Defaulting blank config to `ManifestRewriteStrategyHandler` matches the
other built-ins’ ergonomics.
Related: in `loadStrategyHandler` / `recommendForOneStrategy`, metadata is
fetched before stats, then `hasCurrentSnapshot` compares them. If both
providers briefly return a matching stale snap while the table has moved on,
the check can pass and a rewrite may be submitted against a newer snapshot.
Fetching metadata **after** stats (or re-reading metadata post-stats) would
shrink that TOCTOU window — worth a short comment if you keep the current order.
##########
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) {
Review Comment:
**P2: Dual `initialize` paths invite the wrong caller.**
Single-arg skips when policy `spec_id` is null; two-arg can evaluate with a
caller-supplied `resolvedSpecId` even when the policy omitted `spec_id` (as
long as it doesn’t conflict). Easy to “fix” a collector/IT via the two-arg
overload and ship policies that the CLI/recommender still no-op.
Prefer one production path (require `spec_id` everywhere until cycle handoff
exists), or make the two-arg overload package-private / `@VisibleForTesting`
with a louder contract.
--
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]
