lasdf1234 commented on code in PR #13329: URL: https://github.com/apache/gravitino/pull/13329#discussion_r4069176530
########## 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: Sibling jobs wrap execution in try/catch, print a short error via `System.err` / `printUsage()`, and `System.exit(1)` on failure. Right now `parseArguments`, `parseCustomSparkConfigs`, and `spark.sql()` failures propagate as uncaught exceptions with full stack traces in the job log. Not a functional blocker, but worth aligning for operator experience and consistent non-zero exit behavior. ########## 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: Please use `System.out.println` / `printf` for user-visible results, matching `IcebergExpireSnapshotsJob` and `IcebergRewriteDataFilesJob`. `SparkProcessBuilder` redirects **stdout** to `output.log` and stderr to `error.log`. `LOG.info` here is unlikely to land in `output.log`, but the new docs tell operators to read that file for: ```text Rewrite Manifests Results: Rewritten manifests: 24, Added manifests: 2 ``` Reserve SLF4J for internal/debug logging; keep the result summary on stdout. ########## 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: The stricter parsing (unresolved `{{placeholder}}` handling, unknown/duplicate args, pre-Spark validation) is the right behavior — please keep it. However, having a job-local `parseArguments` diverges from `IcebergRewriteDataFilesJob`, which delegates to `IcebergJobUtils`. Consider lifting this logic into `IcebergJobUtils` (e.g. enhanced parser built on `nullIfUnresolvedPlaceholder`) so future jobs with optional template placeholders can reuse it. This job would then delegate, similar to the rewrite-data-files pattern. ########## 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: These `testParseArgumentsWith*` cases call `IcebergJobUtils.parseArguments()`, not `IcebergRewriteManifestsJob.parseArguments()`. `TestIcebergRewriteDataFilesJob` exercises the job parser (even though it delegates). Please switch these tests to the job parser so duplicate/unknown-arg and placeholder behavior stay covered alongside the generic-utils tests below. -- 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]
