laserninja commented on code in PR #13533: URL: https://github.com/apache/gravitino/pull/13533#discussion_r4226129892
########## 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: Fixed in 026b6add2 with the strict argument parser. After merging current main, the template also declares empty defaults for spec_id and spark_conf using the new {{name:-}} syntax. Local Spark entry-point tests cover omitting both optional values and omitting spec_id with an explicit spark_conf, using the shared template resolver. ########## 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: Fixed in 026b6add2. The job calls requireIcebergSparkRuntime() immediately after Spark creation and before updateStatistics. The throwing variant keeps the existing finally block and try-with-resources cleanup active, so Spark and the updater are closed on failure. ########## 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: Fixed in 026b6add2. The production single-argument initialize path now logs an actionable warning and skips when spec_id is absent instead of aborting recommendation. Added coverage through the real Recommender and documented that empty policy content needs a resolved collection-cycle ID or an explicit spec_id for standalone CLI evaluation. -- 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]
