lasdf1234 commented on code in PR #13533: URL: https://github.com/apache/gravitino/pull/13533#discussion_r4226025344
########## maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergUpdateManifestStatsJob.java: ########## @@ -0,0 +1,223 @@ +/* + * 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.jobs.iceberg; + +import com.google.common.annotations.VisibleForTesting; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import javax.annotation.Nullable; +import org.apache.gravitino.NameIdentifier; +import org.apache.gravitino.job.JobTemplateProvider; +import org.apache.gravitino.job.SparkJobTemplate; +import org.apache.gravitino.maintenance.jobs.BuiltInJob; +import org.apache.gravitino.maintenance.optimizer.api.updater.StatisticsUpdater; +import org.apache.gravitino.maintenance.optimizer.common.IcebergManifestStatistics; +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.IcebergSparkConfigUtils; +import org.apache.gravitino.maintenance.optimizer.common.util.ProviderUtils; +import org.apache.iceberg.ManifestFile; +import org.apache.iceberg.Snapshot; +import org.apache.iceberg.Table; +import org.apache.iceberg.spark.Spark3Util; +import org.apache.spark.sql.SparkSession; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** Built-in job for collecting and atomically merging Iceberg manifest statistics by spec. */ +public class IcebergUpdateManifestStatsJob implements BuiltInJob { + private static final Logger LOG = LoggerFactory.getLogger(IcebergUpdateManifestStatsJob.class); + private static final String NAME = + JobTemplateProvider.BUILTIN_NAME_PREFIX + "iceberg-update-manifest-stats"; + private static final String DEFAULT_STATISTICS_UPDATER = "gravitino-statistics-updater"; + + @Override + public SparkJobTemplate jobTemplate() { + return SparkJobTemplate.builder() + .withName(NAME) + .withComment("Collect manifest count and average size for one Iceberg partition spec") + .withExecutable(resolveExecutable(IcebergUpdateManifestStatsJob.class)) + .withClassName(IcebergUpdateManifestStatsJob.class.getName()) + .withArguments( + Arrays.asList( + "--catalog", + "{{catalog_name}}", + "--table", + "{{table_identifier}}", + "--spec-id", + "{{spec_id}}", + "--updater-options", + "{{updater_options}}", + "--spark-conf", + "{{spark_conf}}")) + .withConfigs(IcebergSparkConfigUtils.buildTemplateSparkConfigs()) + .withCustomFields(Collections.singletonMap(JobTemplateProvider.PROPERTY_VERSION_KEY, "v1")) + .build(); + } + + /** + * Collects and persists manifest statistics using the configured Spark catalog and updater. + * + * @param args catalog, table, optional spec ID, updater options, and Spark configuration + * @throws Exception if collection, persistence, or resource cleanup fails + */ + public static void main(String[] args) throws Exception { + Map<String, String> arguments = IcebergJobUtils.parseArguments(args); + String catalog = arguments.get("catalog"); + String table = arguments.get("table"); + tableIdentifier(catalog, table); + Integer specId = parseSpecId(arguments.get("spec-id")); + Map<String, String> options = + IcebergSparkConfigUtils.parseFlatJsonMap( + arguments.get("updater-options"), "updater-options"); + Map<String, String> properties = buildOptimizerProperties(options); + Map<String, String> sparkConfigs = + IcebergSparkConfigUtils.parseFlatJsonMap(arguments.get("spark-conf"), "spark-conf"); + try (StatisticsUpdater updater = + ProviderUtils.createStatisticsUpdaterInstance( + options.getOrDefault("statistics_updater", DEFAULT_STATISTICS_UPDATER).trim())) { + updater.initialize(new OptimizerEnv(new OptimizerConfig(properties))); + SparkSession.Builder builder = + SparkSession.builder().appName("Gravitino Iceberg Manifest Statistics"); + sparkConfigs.forEach(builder::config); + SparkSession spark = builder.getOrCreate(); + try { + updateStatistics(spark, updater, catalog, table, specId); + } finally { + spark.stop(); + } Review Comment: **Medium:** Missing Iceberg Spark runtime guard after `SparkSession` creation. Sibling built-in Iceberg jobs call `IcebergJobUtils.requireIcebergSparkRuntimeOrExit(spark)` so a misconfigured cluster fails with a clear classpath message instead of a later `ClassNotFoundException` / opaque Spark SQL error from `Spark3Util.loadIcebergTable`. Please add the same check here before `updateStatistics`. ########## maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergUpdateManifestStatsJob.java: ########## @@ -0,0 +1,223 @@ +/* + * 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.jobs.iceberg; + +import com.google.common.annotations.VisibleForTesting; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import javax.annotation.Nullable; +import org.apache.gravitino.NameIdentifier; +import org.apache.gravitino.job.JobTemplateProvider; +import org.apache.gravitino.job.SparkJobTemplate; +import org.apache.gravitino.maintenance.jobs.BuiltInJob; +import org.apache.gravitino.maintenance.optimizer.api.updater.StatisticsUpdater; +import org.apache.gravitino.maintenance.optimizer.common.IcebergManifestStatistics; +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.IcebergSparkConfigUtils; +import org.apache.gravitino.maintenance.optimizer.common.util.ProviderUtils; +import org.apache.iceberg.ManifestFile; +import org.apache.iceberg.Snapshot; +import org.apache.iceberg.Table; +import org.apache.iceberg.spark.Spark3Util; +import org.apache.spark.sql.SparkSession; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** Built-in job for collecting and atomically merging Iceberg manifest statistics by spec. */ +public class IcebergUpdateManifestStatsJob implements BuiltInJob { + private static final Logger LOG = LoggerFactory.getLogger(IcebergUpdateManifestStatsJob.class); + private static final String NAME = + JobTemplateProvider.BUILTIN_NAME_PREFIX + "iceberg-update-manifest-stats"; + private static final String DEFAULT_STATISTICS_UPDATER = "gravitino-statistics-updater"; + + @Override + public SparkJobTemplate jobTemplate() { + return SparkJobTemplate.builder() + .withName(NAME) + .withComment("Collect manifest count and average size for one Iceberg partition spec") + .withExecutable(resolveExecutable(IcebergUpdateManifestStatsJob.class)) + .withClassName(IcebergUpdateManifestStatsJob.class.getName()) + .withArguments( + Arrays.asList( + "--catalog", + "{{catalog_name}}", + "--table", + "{{table_identifier}}", + "--spec-id", + "{{spec_id}}", + "--updater-options", + "{{updater_options}}", + "--spark-conf", + "{{spark_conf}}")) + .withConfigs(IcebergSparkConfigUtils.buildTemplateSparkConfigs()) + .withCustomFields(Collections.singletonMap(JobTemplateProvider.PROPERTY_VERSION_KEY, "v1")) + .build(); + } + + /** + * Collects and persists manifest statistics using the configured Spark catalog and updater. + * + * @param args catalog, table, optional spec ID, updater options, and Spark configuration + * @throws Exception if collection, persistence, or resource cleanup fails + */ + public static void main(String[] args) throws Exception { + Map<String, String> arguments = IcebergJobUtils.parseArguments(args); + String catalog = arguments.get("catalog"); + String table = arguments.get("table"); + tableIdentifier(catalog, table); + Integer specId = parseSpecId(arguments.get("spec-id")); + Map<String, String> options = + IcebergSparkConfigUtils.parseFlatJsonMap( + arguments.get("updater-options"), "updater-options"); + Map<String, String> properties = buildOptimizerProperties(options); + Map<String, String> sparkConfigs = + IcebergSparkConfigUtils.parseFlatJsonMap(arguments.get("spark-conf"), "spark-conf"); Review Comment: **High:** Optional template placeholders are not dropped before parsing. `jobTemplate()` always emits `--spec-id {{spec_id}}` and `--spark-conf {{spark_conf}}`. When those jobConf keys are omitted, the server forwards the literal `{{...}}` strings. This path uses the loose `IcebergJobUtils.parseArguments(String[])` overload, which does **not** call `nullIfUnresolvedPlaceholder`, so: 1. `parseSpecId("{{spec_id}}")` rejects the placeholder (`[0-9]+` check) instead of resolving the table default. 2. `parseFlatJsonMap("{{spark_conf}}", ...)` fails JSON parsing instead of treating spark-conf as absent. `IcebergRewriteManifestsJob` already fixed the same class of bug with the strict `parseArguments(args, supported, required)` overload. Please align this job the same way (or wrap optional values with `nullIfUnresolvedPlaceholder` before `parseSpecId` / `parseFlatJsonMap`). Without that, a normal submit that supplies catalog/table/updater_options but omits optional `spec_id`/`spark_conf` fails before collection starts. ########## maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/recommender/handler/ManifestRewriteStrategyHandler.java: ########## @@ -0,0 +1,170 @@ +/* + * 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.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.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_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()); + Preconditions.checkArgument( + requestedSpec != null, + "Manifest rewrite requires spec_id when evaluating persisted statistics; " + + "for a collection cycle pass the collector's resolved spec ID to initialize"); + initialize(context, requestedSpec); Review Comment: **Medium:** Production `Recommender` only ever calls this single-arg `initialize(context)`, which hard-requires `spec_id` in the policy rules. Creating `system_iceberg_rewrite_manifests` with omitted `spec_id` (documented default / empty content) succeeds validation, but `Recommender.loadStrategyHandler` then throws `IllegalArgumentException` and aborts recommendation for that identifier rather than skipping. The two-arg `initialize(context, resolvedSpecId)` path is only exercised from tests / direct callers; nothing in the optimizer wires the collector's resolved spec into evaluation yet. If CLI/optimizer evaluation is the supported path in this PR, consider either: - rejecting omitted `spec_id` at policy `validate()` time with the same message, or - catching/skipping this case in `Recommender` so one misconfigured policy does not look like a fatal handler failure. If the collection-cycle API is intentionally deferred, a short note that empty content is not CLI-evaluable helps; the create-time accept / evaluate-time throw split is easy to miss. -- 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]
