laserninja commented on code in PR #13329: URL: https://github.com/apache/gravitino/pull/13329#discussion_r4073914590
########## maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteManifestsJob.java: ########## @@ -0,0 +1,264 @@ +/* + * 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 java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import javax.annotation.Nullable; +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.common.util.IcebergSparkConfigUtils; +import org.apache.spark.sql.Row; +import org.apache.spark.sql.SparkSession; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Built-in job for rewriting Iceberg table manifest files. + * + * <p>This job leverages Iceberg's RewriteManifestsProcedure to consolidate small manifest files and + * cluster manifest entries within an existing partition spec to improve scan planning. + */ +public class IcebergRewriteManifestsJob implements BuiltInJob { + + private static final Logger LOG = LoggerFactory.getLogger(IcebergRewriteManifestsJob.class); + + private static final String NAME = + JobTemplateProvider.BUILTIN_NAME_PREFIX + "iceberg-rewrite-manifests"; + private static final String VERSION = "v1"; + + @Override + public SparkJobTemplate jobTemplate() { + return SparkJobTemplate.builder() + .withName(NAME) + .withComment( + "Built-in Iceberg rewrite manifests job template for scan planning optimization") + .withExecutable(resolveExecutable(IcebergRewriteManifestsJob.class)) + .withClassName(IcebergRewriteManifestsJob.class.getName()) + .withArguments(buildArguments()) + .withConfigs(buildSparkConfigs()) + .withCustomFields( + Collections.singletonMap(JobTemplateProvider.PROPERTY_VERSION_KEY, VERSION)) + .build(); + } + + /** + * Main entry point for the rewrite manifests job. + * + * <p>Uses named arguments for flexibility: + * + * <ul> + * <li>--catalog <catalog_name> Required. Iceberg catalog name. + * <li>--table <table_identifier> Required. Table name (db.table) + * <li>--use-caching <boolean> Optional. Whether to cache the table metadata in Spark + * while rewriting (default: Iceberg's own default) + * <li>--spec-id <int> Optional. Rewrite manifests belonging to this partition spec ID + * (default: the table's current spec) + * <li>--spark-conf <spark_conf_json> Optional. JSON map of custom Spark configurations + * </ul> + * + * <p><b>Important Notes on Special Characters:</b> + * + * <ul> + * <li><b>Via Gravitino API:</b> Pass values as-is without shell escaping. Gravitino handles + * escaping internally via ProcessBuilder. + * <li><b>Via Command Line:</b> Use shell quoting for values containing whitespace. + * </ul> + * + * <p>Example via command line: --catalog iceberg_catalog --table db.sample --use-caching false + * + * <p>Example via Gravitino API: + * + * <pre>{@code + * Map<String, String> jobConf = new HashMap<>(); + * jobConf.put("catalog_name", "iceberg_catalog"); + * jobConf.put("table_identifier", "db.sample"); + * jobConf.put("use_caching", "false"); + * metalake.runJob("builtin-iceberg-rewrite-manifests", jobConf); + * }</pre> + * + * @param args named job arguments + */ + public static void main(String[] args) { + Map<String, String> argMap = parseArguments(args); + String sql = + buildProcedureCall( + argMap.get("catalog"), + argMap.get("table"), + argMap.get("use-caching"), + argMap.get("spec-id")); + Map<String, String> configs = IcebergJobUtils.parseCustomSparkConfigs(argMap.get("spark-conf")); + SparkSession.Builder builder = + SparkSession.builder().appName("Gravitino Built-in Iceberg Rewrite Manifests"); + configs.forEach(builder::config); + + try (SparkSession spark = builder.getOrCreate()) { + IcebergJobUtils.requireIcebergSparkRuntime(); + List<Row> results = spark.sql(sql).collectAsList(); + if (!results.isEmpty()) { + Row result = results.get(0); + LOG.info( Review Comment: Fixed in 8961a7a66. The result summary now uses System.out.printf and the completion message uses stdout, so both reach output.log. The Spark-backed tests capture stdout and assert that the result summary is present; this assertion failed before the fix. ########## maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteManifestsJob.java: ########## @@ -0,0 +1,264 @@ +/* + * 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 java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import javax.annotation.Nullable; +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.common.util.IcebergSparkConfigUtils; +import org.apache.spark.sql.Row; +import org.apache.spark.sql.SparkSession; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Built-in job for rewriting Iceberg table manifest files. + * + * <p>This job leverages Iceberg's RewriteManifestsProcedure to consolidate small manifest files and + * cluster manifest entries within an existing partition spec to improve scan planning. + */ +public class IcebergRewriteManifestsJob implements BuiltInJob { + + private static final Logger LOG = LoggerFactory.getLogger(IcebergRewriteManifestsJob.class); + + private static final String NAME = + JobTemplateProvider.BUILTIN_NAME_PREFIX + "iceberg-rewrite-manifests"; + private static final String VERSION = "v1"; + + @Override + public SparkJobTemplate jobTemplate() { + return SparkJobTemplate.builder() + .withName(NAME) + .withComment( + "Built-in Iceberg rewrite manifests job template for scan planning optimization") + .withExecutable(resolveExecutable(IcebergRewriteManifestsJob.class)) + .withClassName(IcebergRewriteManifestsJob.class.getName()) + .withArguments(buildArguments()) + .withConfigs(buildSparkConfigs()) + .withCustomFields( + Collections.singletonMap(JobTemplateProvider.PROPERTY_VERSION_KEY, VERSION)) + .build(); + } + + /** + * Main entry point for the rewrite manifests job. + * + * <p>Uses named arguments for flexibility: + * + * <ul> + * <li>--catalog <catalog_name> Required. Iceberg catalog name. + * <li>--table <table_identifier> Required. Table name (db.table) + * <li>--use-caching <boolean> Optional. Whether to cache the table metadata in Spark + * while rewriting (default: Iceberg's own default) + * <li>--spec-id <int> Optional. Rewrite manifests belonging to this partition spec ID + * (default: the table's current spec) + * <li>--spark-conf <spark_conf_json> Optional. JSON map of custom Spark configurations + * </ul> + * + * <p><b>Important Notes on Special Characters:</b> + * + * <ul> + * <li><b>Via Gravitino API:</b> Pass values as-is without shell escaping. Gravitino handles + * escaping internally via ProcessBuilder. + * <li><b>Via Command Line:</b> Use shell quoting for values containing whitespace. + * </ul> + * + * <p>Example via command line: --catalog iceberg_catalog --table db.sample --use-caching false + * + * <p>Example via Gravitino API: + * + * <pre>{@code + * Map<String, String> jobConf = new HashMap<>(); + * jobConf.put("catalog_name", "iceberg_catalog"); + * jobConf.put("table_identifier", "db.sample"); + * jobConf.put("use_caching", "false"); + * metalake.runJob("builtin-iceberg-rewrite-manifests", jobConf); + * }</pre> + * + * @param args named job arguments + */ + public static void main(String[] args) { Review Comment: Fixed in 8961a7a66. Failures now produce a concise error and usage message on stderr, with main exiting 1 after Spark cleanup. Process-level checks verified exit 1 without an uncaught stack trace for missing arguments and malformed Spark configuration. The unknown-spec Spark test also verifies failure reporting, session cleanup, and unchanged table metadata. ########## maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteManifestsJob.java: ########## @@ -0,0 +1,264 @@ +/* + * 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 java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import javax.annotation.Nullable; +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.common.util.IcebergSparkConfigUtils; +import org.apache.spark.sql.Row; +import org.apache.spark.sql.SparkSession; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Built-in job for rewriting Iceberg table manifest files. + * + * <p>This job leverages Iceberg's RewriteManifestsProcedure to consolidate small manifest files and + * cluster manifest entries within an existing partition spec to improve scan planning. + */ +public class IcebergRewriteManifestsJob implements BuiltInJob { + + private static final Logger LOG = LoggerFactory.getLogger(IcebergRewriteManifestsJob.class); + + private static final String NAME = + JobTemplateProvider.BUILTIN_NAME_PREFIX + "iceberg-rewrite-manifests"; + private static final String VERSION = "v1"; + + @Override + public SparkJobTemplate jobTemplate() { + return SparkJobTemplate.builder() + .withName(NAME) + .withComment( + "Built-in Iceberg rewrite manifests job template for scan planning optimization") + .withExecutable(resolveExecutable(IcebergRewriteManifestsJob.class)) + .withClassName(IcebergRewriteManifestsJob.class.getName()) + .withArguments(buildArguments()) + .withConfigs(buildSparkConfigs()) + .withCustomFields( + Collections.singletonMap(JobTemplateProvider.PROPERTY_VERSION_KEY, VERSION)) + .build(); + } + + /** + * Main entry point for the rewrite manifests job. + * + * <p>Uses named arguments for flexibility: + * + * <ul> + * <li>--catalog <catalog_name> Required. Iceberg catalog name. + * <li>--table <table_identifier> Required. Table name (db.table) + * <li>--use-caching <boolean> Optional. Whether to cache the table metadata in Spark + * while rewriting (default: Iceberg's own default) + * <li>--spec-id <int> Optional. Rewrite manifests belonging to this partition spec ID + * (default: the table's current spec) + * <li>--spark-conf <spark_conf_json> Optional. JSON map of custom Spark configurations + * </ul> + * + * <p><b>Important Notes on Special Characters:</b> + * + * <ul> + * <li><b>Via Gravitino API:</b> Pass values as-is without shell escaping. Gravitino handles + * escaping internally via ProcessBuilder. + * <li><b>Via Command Line:</b> Use shell quoting for values containing whitespace. + * </ul> + * + * <p>Example via command line: --catalog iceberg_catalog --table db.sample --use-caching false + * + * <p>Example via Gravitino API: + * + * <pre>{@code + * Map<String, String> jobConf = new HashMap<>(); + * jobConf.put("catalog_name", "iceberg_catalog"); + * jobConf.put("table_identifier", "db.sample"); + * jobConf.put("use_caching", "false"); + * metalake.runJob("builtin-iceberg-rewrite-manifests", jobConf); + * }</pre> + * + * @param args named job arguments + */ + public static void main(String[] args) { + Map<String, String> argMap = parseArguments(args); + String sql = + buildProcedureCall( + argMap.get("catalog"), + argMap.get("table"), + argMap.get("use-caching"), + argMap.get("spec-id")); + Map<String, String> configs = IcebergJobUtils.parseCustomSparkConfigs(argMap.get("spark-conf")); + SparkSession.Builder builder = + SparkSession.builder().appName("Gravitino Built-in Iceberg Rewrite Manifests"); + configs.forEach(builder::config); + + try (SparkSession spark = builder.getOrCreate()) { + IcebergJobUtils.requireIcebergSparkRuntime(); + List<Row> results = spark.sql(sql).collectAsList(); + if (!results.isEmpty()) { + Row result = results.get(0); + LOG.info( + "Rewrite Manifests Results: Rewritten manifests: {}, Added manifests: {}", + ((Number) result.get(0)).longValue(), + ((Number) result.get(1)).longValue()); + } + LOG.info("Rewrite manifests job completed successfully"); + } + } + + static Map<String, String> parseArguments(String[] args) { Review Comment: Fixed in 8961a7a66. Moved strict parsing into an IcebergJobUtils.parseArguments overload that accepts supported and required argument names and uses nullIfUnresolvedPlaceholder. The job delegates to it and retains its boolean/spec validation before Spark starts. The existing overload keeps its behavior for other jobs; added shared-parser tests for normalization and invalid input. ########## maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteManifestsJob.java: ########## @@ -0,0 +1,415 @@ +/* + * 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 static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.Map; +import org.apache.gravitino.job.JobTemplateProvider; +import org.apache.gravitino.job.SparkJobTemplate; +import org.junit.jupiter.api.Test; + +public class TestIcebergRewriteManifestsJob { + + @Test + public void testInvalidEntryArgumentsFailBeforeSparkStarts() { + String[][] invalid = { + {}, + {"--catalog", "cat", "--table"}, + {"--catalog", "{{catalog_name}}", "--table", "db.t"}, + {"--catalog", " ", "--table", "db.t"}, + {"--catalog", "cat", "--table", "db.t", "--use-caching", "yes"}, + {"--catalog", "cat", "--table", "db.t", "--spec-id", "-1"}, + {"--catalog", "cat", "--table", "db.t", "--spark-conf", "not-json"}, + {"--catalog", "cat", "--table", "db.t", "--unknown", "value"}, + {"--catalog", "cat", "--table", "db.t", "--catalog", "other"} + }; + for (String[] args : invalid) { + assertThrows(IllegalArgumentException.class, () -> IcebergRewriteManifestsJob.main(args)); + } + } + + @Test + public void testEntryArgumentsOmitEmptyAndUnresolvedOptionals() { + Map<String, String> args = + IcebergRewriteManifestsJob.parseArguments( + new String[] { + "--catalog", + "cat", + "--table", + "db.t", + "--use-caching", + "{{use_caching}}", + "--spec-id", + "", + "--spark-conf", + " " + }); + assertNull(args.get("use-caching")); + assertNull(args.get("spec-id")); + assertNull(args.get("spark-conf")); + } + + @Test + public void testJobTemplateHasCorrectName() { + IcebergRewriteManifestsJob job = new IcebergRewriteManifestsJob(); + SparkJobTemplate template = job.jobTemplate(); + + assertNotNull(template); + assertEquals("builtin-iceberg-rewrite-manifests", template.name()); + } + + @Test + public void testJobTemplateHasComment() { + IcebergRewriteManifestsJob job = new IcebergRewriteManifestsJob(); + SparkJobTemplate template = job.jobTemplate(); + + assertNotNull(template.comment()); + assertFalse(template.comment().trim().isEmpty()); + assertTrue(template.comment().contains("Iceberg")); + assertTrue(template.comment().contains("manifests")); + } + + @Test + public void testJobTemplateHasExecutable() { + IcebergRewriteManifestsJob job = new IcebergRewriteManifestsJob(); + SparkJobTemplate template = job.jobTemplate(); + + assertNotNull(template.executable()); + assertFalse(template.executable().trim().isEmpty()); + } + + @Test + public void testJobTemplateHasClassName() { + IcebergRewriteManifestsJob job = new IcebergRewriteManifestsJob(); + SparkJobTemplate template = job.jobTemplate(); + + assertNotNull(template.className()); + assertEquals(IcebergRewriteManifestsJob.class.getName(), template.className()); + } + + @Test + public void testJobTemplateHasArguments() { + IcebergRewriteManifestsJob job = new IcebergRewriteManifestsJob(); + SparkJobTemplate template = job.jobTemplate(); + + assertNotNull(template.arguments()); + assertEquals(10, template.arguments().size()); // 5 flags * 2 (flag + value) + + // Verify all expected arguments are present + assertTrue(template.arguments().contains("--catalog")); + assertTrue(template.arguments().contains("{{catalog_name}}")); + assertTrue(template.arguments().contains("--table")); + assertTrue(template.arguments().contains("{{table_identifier}}")); + assertTrue(template.arguments().contains("--use-caching")); + assertTrue(template.arguments().contains("{{use_caching}}")); + assertTrue(template.arguments().contains("--spec-id")); + assertTrue(template.arguments().contains("{{spec_id}}")); + assertTrue(template.arguments().contains("--spark-conf")); + assertTrue(template.arguments().contains("{{spark_conf}}")); + } + + @Test + public void testJobTemplateHasSparkConfigs() { + IcebergRewriteManifestsJob job = new IcebergRewriteManifestsJob(); + SparkJobTemplate template = job.jobTemplate(); + + Map<String, String> configs = template.configs(); + assertNotNull(configs); + assertFalse(configs.isEmpty()); + + // Verify Spark runtime configs + assertTrue(configs.containsKey("spark.master")); + assertTrue(configs.containsKey("spark.executor.instances")); + assertTrue(configs.containsKey("spark.executor.cores")); + assertTrue(configs.containsKey("spark.executor.memory")); + assertTrue(configs.containsKey("spark.driver.memory")); + + // Verify Iceberg catalog configs + assertTrue(configs.containsKey("spark.sql.catalog.{{catalog_name}}")); + assertTrue(configs.containsKey("spark.sql.catalog.{{catalog_name}}.type")); + assertTrue(configs.containsKey("spark.sql.catalog.{{catalog_name}}.uri")); + assertTrue(configs.containsKey("spark.sql.catalog.{{catalog_name}}.warehouse")); + } + + @Test + public void testJobTemplateHasVersion() { + IcebergRewriteManifestsJob job = new IcebergRewriteManifestsJob(); + SparkJobTemplate template = job.jobTemplate(); + + Map<String, String> customFields = template.customFields(); + assertNotNull(customFields); + assertTrue(customFields.containsKey(JobTemplateProvider.PROPERTY_VERSION_KEY)); + + String version = customFields.get(JobTemplateProvider.PROPERTY_VERSION_KEY); + assertEquals("v1", version); + assertTrue(version.matches(JobTemplateProvider.VERSION_VALUE_PATTERN)); + } + + @Test + public void testJobTemplateNameMatchesBuiltInPattern() { + IcebergRewriteManifestsJob job = new IcebergRewriteManifestsJob(); + SparkJobTemplate template = job.jobTemplate(); + + assertTrue(template.name().matches(JobTemplateProvider.BUILTIN_NAME_PATTERN)); + assertTrue(template.name().startsWith(JobTemplateProvider.BUILTIN_NAME_PREFIX)); + } + + // Test parseArguments method + + @Test + public void testParseArgumentsWithAllRequired() { + String[] args = {"--catalog", "iceberg_prod", "--table", "db.sample"}; + Map<String, String> result = IcebergJobUtils.parseArguments(args); Review Comment: Fixed in 8961a7a66. The argument-parsing tests now call IcebergRewriteManifestsJob.parseArguments, with assertions for normalized optional values and rejection of duplicate, unknown, missing, and unresolved required arguments. Shared utility tests remain separate. All 183 jobs-module tests passed, including the three Spark-backed manifest tests. -- 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]
