From d394d2a6568672e1d6c9720212ca562dffdaf3df Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C4=B0brahim=20ERB=C4=B0LEN?= Date: Mon, 25 May 2026 20:46:39 +0300 Subject: [PATCH 1/7] Add builtin-iceberg-rewrite-manifests job --- .../jobs/BuiltInJobTemplateProvider.java | 2 + .../iceberg/IcebergRewriteManifestsJob.java | 234 ++++++++++++++++++ 2 files changed, 236 insertions(+) create mode 100644 maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteManifestsJob.java diff --git a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/BuiltInJobTemplateProvider.java b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/BuiltInJobTemplateProvider.java index 3053c02d96b..40ecafe6335 100644 --- a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/BuiltInJobTemplateProvider.java +++ b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/BuiltInJobTemplateProvider.java @@ -27,6 +27,7 @@ import org.apache.gravitino.job.JobTemplateProvider; import org.apache.gravitino.maintenance.jobs.iceberg.IcebergExpireSnapshotsJob; import org.apache.gravitino.maintenance.jobs.iceberg.IcebergRewriteDataFilesJob; +import org.apache.gravitino.maintenance.jobs.iceberg.IcebergRewriteManifestsJob; import org.apache.gravitino.maintenance.jobs.iceberg.IcebergUpdateStatsAndMetricsJob; import org.apache.gravitino.maintenance.jobs.spark.SparkPiJob; import org.slf4j.Logger; @@ -46,6 +47,7 @@ public class BuiltInJobTemplateProvider implements JobTemplateProvider { ImmutableList.of( new SparkPiJob(), new IcebergRewriteDataFilesJob(), + new IcebergRewriteManifestsJob(), new IcebergUpdateStatsAndMetricsJob(), new IcebergExpireSnapshotsJob()); diff --git a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteManifestsJob.java b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteManifestsJob.java new file mode 100644 index 00000000000..0caa662fc5a --- /dev/null +++ b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteManifestsJob.java @@ -0,0 +1,234 @@ +/* + * 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.List; +import java.util.Map; +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; + +/** + * Built-in job for rewriting Iceberg table manifest files. + * + *

This job leverages Iceberg's RewriteManifestsProcedure to consolidate small manifest files + * and rewrite manifests with improved partition specs, which improves scan planning performance. + */ +public class IcebergRewriteManifestsJob implements BuiltInJob { + + 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. + * + *

Uses named arguments for flexibility: + * + *

+ * + *

Example via Gravitino API: + * + * 

{@code
+   * Map jobConf = new HashMap<>();
+   * jobConf.put("catalog_name", "iceberg_catalog");
+   * jobConf.put("table_identifier", "db.sample");
+   * jobConf.put("use_caching", "true");
+   * metalake.runJob("builtin-iceberg-rewrite-manifests", jobConf);
+   * }
+ */ + public static void main(String[] args) { + if (args.length < 4) { + printUsage(); + System.exit(1); + } + + Map argMap = parseArguments(args); + + String catalogName = argMap.get("catalog"); + String tableIdentifier = argMap.get("table"); + + if (catalogName == null || tableIdentifier == null) { + System.err.println("Error: --catalog and --table are required arguments"); + printUsage(); + System.exit(1); + } + + String useCaching = argMap.get("use-caching"); + String sparkConfJson = argMap.get("spark-conf"); + + SparkSession.Builder sparkBuilder = + SparkSession.builder().appName("Gravitino Built-in Iceberg Rewrite Manifests"); + + if (sparkConfJson != null && !sparkConfJson.isEmpty()) { + try { + Map customConfigs = + IcebergRewriteDataFilesJob.parseCustomSparkConfigs(sparkConfJson); + for (Map.Entry entry : customConfigs.entrySet()) { + sparkBuilder.config(entry.getKey(), entry.getValue()); + } + System.out.println("Applied custom Spark configurations: " + customConfigs); + } catch (IllegalArgumentException e) { + System.err.println("Error: " + e.getMessage()); + printUsage(); + System.exit(1); + } + } + + SparkSession spark = sparkBuilder.getOrCreate(); + + try { + String sql = buildProcedureCall(catalogName, tableIdentifier, useCaching); + System.out.println("Executing Iceberg rewrite_manifests procedure: " + sql); + + Row[] results = (Row[]) spark.sql(sql).collect(); + + if (results.length > 0) { + Row result = results[0]; + System.out.printf( + "Rewrite Manifests Results:%n" + + " Rewritten manifests: %d%n" + + " Added manifests: %d%n", + result.getInt(0), result.getInt(1)); + } + + System.out.println("Rewrite manifests job completed successfully"); + } catch (Exception e) { + System.err.println("Error executing rewrite manifests job: " + e.getMessage()); + e.printStackTrace(); + System.exit(1); + } finally { + spark.stop(); + } + } + + /** + * Build the SQL CALL statement for the rewrite_manifests procedure. + * + * @param catalogName Iceberg catalog name + * @param tableIdentifier Fully qualified table name + * @param useCaching Whether to use caching during rewrite + * @return SQL CALL statement + */ + static String buildProcedureCall( + String catalogName, String tableIdentifier, String useCaching) { + StringBuilder sql = new StringBuilder(); + sql.append("CALL ") + .append(IcebergRewriteDataFilesJob.escapeSqlIdentifier(catalogName)) + .append(".system.rewrite_manifests("); + sql.append("table => '") + .append(IcebergRewriteDataFilesJob.escapeSqlString(tableIdentifier)) + .append("'"); + + if (useCaching != null && !useCaching.isEmpty()) { + sql.append(", use_caching => ").append(Boolean.parseBoolean(useCaching)); + } + + sql.append(")"); + return sql.toString(); + } + + /** + * Parse command line arguments in --key value format. + * + * @param args command line arguments + * @return map of argument names to values + */ + static Map parseArguments(String[] args) { + Map argMap = new HashMap<>(); + for (int i = 0; i < args.length; i++) { + if (args[i].startsWith("--")) { + String key = args[i].substring(2); + if (i + 1 < args.length && !args[i + 1].startsWith("--")) { + String value = args[i + 1]; + if (value != null && !value.trim().isEmpty()) { + argMap.put(key, value); + } + i++; + } else { + System.err.println("Warning: Flag " + args[i] + " has no value, ignoring"); + } + } + } + return argMap; + } + + private static void printUsage() { + System.err.println( + "Usage: IcebergRewriteManifestsJob [OPTIONS]\n" + + "\n" + + "Required Options:\n" + + " --catalog Iceberg catalog name registered in Spark\n" + + " --table Fully qualified table name (e.g., db.table_name)\n" + + "\n" + + "Optional Options:\n" + + " --use-caching Whether to use caching during rewrite (default: true)\n" + + " --spark-conf JSON map of custom Spark configurations\n" + + " Example: '{\"spark.sql.shuffle.partitions\":\"200\"}'\n" + + "\n" + + "Examples:\n" + + " # Basic rewrite\n" + + " --catalog iceberg_prod --table db.sample\n" + + "\n" + + " # Without caching\n" + + " --catalog iceberg_prod --table db.sample --use-caching false\n"); + } + + private static List buildArguments() { + return Arrays.asList( + "--catalog", + "{{catalog_name}}", + "--table", + "{{table_identifier}}", + "--use-caching", + "{{use_caching}}", + "--spark-conf", + "{{spark_conf}}"); + } + + private static Map buildSparkConfigs() { + return IcebergSparkConfigUtils.buildTemplateSparkConfigs(); + } +} \ No newline at end of file From 359f83d56ffc15ae7cee7fcd9c440b410be91d8e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C4=B0brahim=20ERB=C4=B0LEN?= Date: Mon, 25 May 2026 21:03:57 +0300 Subject: [PATCH 2/7] feat(maintenance): Add builtin-iceberg-rewrite-manifests job and unit tests --- .../TestIcebergRewriteManifestsJob.java | 254 ++++++++++++++++++ 1 file changed, 254 insertions(+) create mode 100644 maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteManifestsJob.java diff --git a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteManifestsJob.java b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteManifestsJob.java new file mode 100644 index 00000000000..30e19357ac9 --- /dev/null +++ b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteManifestsJob.java @@ -0,0 +1,254 @@ +/* + * 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.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +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 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 testJobTemplateHasMainClass() { + 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(8, template.arguments().size()); // 4 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("--spark-conf")); + assertTrue(template.arguments().contains("{{spark_conf}}")); + } + + @Test + public void testJobTemplateHasSparkConfigs() { + IcebergRewriteManifestsJob job = new IcebergRewriteManifestsJob(); + SparkJobTemplate template = job.jobTemplate(); + + Map 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 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 result = IcebergRewriteManifestsJob.parseArguments(args); + + assertEquals(2, result.size()); + assertEquals("iceberg_prod", result.get("catalog")); + assertEquals("db.sample", result.get("table")); + } + + @Test + public void testParseArgumentsWithOptional() { + String[] args = { + "--catalog", "iceberg_prod", + "--table", "db.sample", + "--use-caching", "false" + }; + Map result = IcebergRewriteManifestsJob.parseArguments(args); + + assertEquals(3, result.size()); + assertEquals("iceberg_prod", result.get("catalog")); + assertEquals("db.sample", result.get("table")); + assertEquals("false", result.get("use-caching")); + } + + @Test + public void testParseArgumentsWithEmptyValues() { + String[] args = {"--catalog", "iceberg_prod", "--table", "db.sample", "--use-caching", ""}; + Map result = IcebergRewriteManifestsJob.parseArguments(args); + + // Empty values should be ignored + assertEquals(2, result.size()); + assertEquals("iceberg_prod", result.get("catalog")); + assertEquals("db.sample", result.get("table")); + assertFalse(result.containsKey("use-caching")); + } + + @Test + public void testParseArgumentsWithMissingValues() { + String[] args = {"--catalog", "iceberg_prod", "--table"}; + Map result = IcebergRewriteManifestsJob.parseArguments(args); + + // Only catalog should be parsed, table has no value + assertEquals(1, result.size()); + assertEquals("iceberg_prod", result.get("catalog")); + assertFalse(result.containsKey("table")); + } + + @Test + public void testParseArgumentsOrderIndependent() { + String[] args1 = {"--catalog", "cat1", "--table", "tbl1", "--use-caching", "true"}; + String[] args2 = {"--use-caching", "true", "--table", "tbl1", "--catalog", "cat1"}; + + Map result1 = IcebergRewriteManifestsJob.parseArguments(args1); + Map result2 = IcebergRewriteManifestsJob.parseArguments(args2); + + assertEquals(result1, result2); + } + + // Test buildProcedureCall method + + @Test + public void testBuildProcedureCallMinimal() { + String sql = + IcebergRewriteManifestsJob.buildProcedureCall("iceberg_prod", "db.sample", null); + + assertEquals("CALL iceberg_prod.system.rewrite_manifests(table => 'db.sample')", sql); + } + + @Test + public void testBuildProcedureCallWithCachingTrue() { + String sql = + IcebergRewriteManifestsJob.buildProcedureCall("iceberg_prod", "db.sample", "true"); + + assertEquals( + "CALL iceberg_prod.system.rewrite_manifests(table => 'db.sample', use_caching => true)", + sql); + } + + @Test + public void testBuildProcedureCallWithCachingFalse() { + String sql = + IcebergRewriteManifestsJob.buildProcedureCall("iceberg_prod", "db.sample", "false"); + + assertEquals( + "CALL iceberg_prod.system.rewrite_manifests(table => 'db.sample', use_caching => false)", + sql); + } + + @Test + public void testBuildProcedureCallWithEmptyCaching() { + String sql = + IcebergRewriteManifestsJob.buildProcedureCall("iceberg_prod", "db.sample", ""); + + assertEquals("CALL iceberg_prod.system.rewrite_manifests(table => 'db.sample')", sql); + } + + @Test + public void testBuildProcedureCallWithSqlInjectionAttempt() { + // Test SQL injection attempt in table name + String maliciousTable = "db.table' OR '1'='1"; + String sql = + IcebergRewriteManifestsJob.buildProcedureCall("iceberg_catalog", maliciousTable, null); + + // Verify single quotes are escaped (becomes '') + assertTrue(sql.contains("db.table'' OR ''1''=''1")); + assertFalse(sql.contains("' OR '1'='1")); + + // Test SQL injection attempt in catalog name + String maliciousCatalog = "catalog`; DROP TABLE users; --"; + sql = + IcebergRewriteManifestsJob.buildProcedureCall(maliciousCatalog, "db.table", null); + + // Verify backticks are escaped + assertTrue(sql.contains("catalog``; DROP TABLE users; --")); + } +} \ No newline at end of file From c5ff4d62125bb85532a6eb59853b930a3ba0713b Mon Sep 17 00:00:00 2001 From: Akshay Thorat Date: Sun, 6 Sep 2026 19:08:42 -0700 Subject: [PATCH 3/7] fix(maintenance): address review feedback on rewrite-manifests job MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Rebases the job onto current main and resolves the review comments left on apache/gravitino#11216: - Fix the misplaced closing parenthesis in BuiltInJobTemplateProvider that left the new job outside ImmutableList.of(...) and broke compilation, and restore alphabetical import ordering. - Use IcebergJobUtils for SQL escaping, argument parsing and Spark config parsing instead of reaching into IcebergRewriteDataFilesJob, and drop the duplicated parseArguments implementation. - Drop unresolved {{...}} template placeholders for optional arguments. The server forwards them verbatim when a jobConf key is absent, so use_caching => {{use_caching}} silently parsed to false instead of leaving Iceberg's default in place. - Reject a use-caching value that is neither true nor false rather than letting Boolean.parseBoolean turn a typo into false. - Read the procedure output through collectAsList and Number so the job does not depend on whether the runtime Iceberg version types the manifest counts as int or long. The Iceberg jar is a compileOnly dependency supplied by the cluster, so that width can differ from what we compile against. - Update the SQL assertions for backtick-quoted catalog identifiers, which is what escapeSqlIdentifier now produces, and rename testJobTemplateHasMainClass to match the property it asserts on. - Document the job in the table maintenance CLI reference alongside the other three built-in templates. Co-authored-by: İbrahim ERBİLEN --- .../optimizer-cli-reference.md | 80 ++++++++++- .../jobs/iceberg/IcebergJobUtils.java | 30 ++++- .../iceberg/IcebergRewriteManifestsJob.java | 125 +++++++++++------- .../TestIcebergRewriteManifestsJob.java | 116 +++++++++++----- 4 files changed, 271 insertions(+), 80 deletions(-) diff --git a/docs/table-maintenance-service/optimizer-cli-reference.md b/docs/table-maintenance-service/optimizer-cli-reference.md index fcfa9130a09..25603990ea8 100644 --- a/docs/table-maintenance-service/optimizer-cli-reference.md +++ b/docs/table-maintenance-service/optimizer-cli-reference.md @@ -234,13 +234,14 @@ EvaluationResult{scopeType=TABLE, identifier=rest_catalog.db.t1, partitionPath=< ## Built-in Job Templates -Three job templates ship with the service, and they are complementary rather than alternatives. A full maintenance pass collects statistics, compacts data files, and then expires the snapshot history that compaction just created. +Four job templates ship with the service, and they are complementary rather than alternatives. A full maintenance pass collects statistics, compacts data files, expires the snapshot history that compaction just created, and consolidates the manifests left behind. | Job template | What it does | |---------------------------------------|-------------------------------------------| | `builtin-iceberg-update-stats` | Collects file statistics and metrics | | `builtin-iceberg-rewrite-data-files` | Compacts small data files | | `builtin-iceberg-expire-snapshots` | Removes old snapshot metadata | +| `builtin-iceberg-rewrite-manifests` | Consolidates small manifest files | Each can be submitted directly over REST, and the first two are also what the policy-driven workflow submits on your behalf. See [Quick Start](./optimizer.md#walkthrough) for the policy-driven path. @@ -340,6 +341,83 @@ Expire Snapshots Results: Deleted manifest lists: 3 ``` +## Rewrite Manifests + +`builtin-iceberg-rewrite-manifests` consolidates a table's manifest files. Frequent commits leave behind many small manifests, and scan planning has to open every one of them to decide which data files a filter matches. Rewriting them into fewer, larger manifests aligned with the current partition spec cuts that planning cost. + +This complements `builtin-iceberg-rewrite-data-files`: that job improves the data file layout, this one improves the metadata that points at it. + +The job calls Iceberg's `rewrite_manifests` stored procedure through Spark SQL. + +| Property | Value | +|-------------|------------------------------------------------------------------------------| +| Name | `builtin-iceberg-rewrite-manifests` | +| Type | Spark | +| Version | `v1` | +| Main class | `org.apache.gravitino.maintenance.jobs.iceberg.IcebergRewriteManifestsJob` | + +### Parameters + +`catalog_name` and `table_identifier` are required. The rest are optional. + +| Key | Description | Default | +|--------------------|--------------------------------------------------------------------|-----------------------------| +| `catalog_name` | Iceberg catalog name as registered in Spark | Required | +| `table_identifier` | Fully qualified table name, such as `db.sample` | Required | +| `use_caching` | Caches table metadata in Spark while rewriting; `true` or `false` | `true` (Iceberg default) | +| `spark_conf` | JSON map of Spark configuration | None | + +Leave `use_caching` unset unless the driver is memory constrained. Caching keeps the table metadata in Spark for the duration of the rewrite, which is faster but holds more memory. + +### Submitting the Job + +```bash +curl -X POST -H "Accept: application/vnd.gravitino.v1+json" \ + -H "Content-Type: application/json" \ + -d '{ + "jobTemplateName": "builtin-iceberg-rewrite-manifests", + "jobConf": { + "catalog_name": "rest_catalog", + "table_identifier": "db.t1", + "spark_master": "local[2]", + "spark_executor_instances": "1", + "spark_executor_cores": "1", + "spark_executor_memory": "1g", + "spark_driver_memory": "1g", + "catalog_type": "rest", + "catalog_uri": "http://localhost:9001/iceberg", + "warehouse_location": "" + } + }' \ + http://localhost:8090/api/metalakes/test/jobs +``` + +The job builds this statement, including `use_caching` only when you supply it: + +```sql +CALL `rest_catalog`.system.rewrite_manifests( + table => 'db.t1', + use_caching => false +) +``` + +### Verifying the Result + +```bash +curl -sS "http://localhost:8090/api/metalakes/test/jobs/{job_id}" | jq '.job.state' +cat /tmp/gravitino/jobs/staging/test/builtin-iceberg-rewrite-manifests/{job_id}/stdout.log +``` + +A successful run reports its state as `SUCCEEDED` and logs how many manifests it replaced: + +```text +Rewrite Manifests Results: + Rewritten manifests: 24 + Added manifests: 2 +``` + +Both counts at zero means the table's manifests were already consolidated, which is a successful no-op rather than a failure. + ## Related - [Table Maintenance Service](./optimizer.md) for the concepts and the walkthrough diff --git a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergJobUtils.java b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergJobUtils.java index 19cdc2f051f..9f46a096750 100644 --- a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergJobUtils.java +++ b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergJobUtils.java @@ -22,17 +22,43 @@ import com.fasterxml.jackson.databind.ObjectMapper; import java.util.HashMap; import java.util.Map; +import java.util.regex.Pattern; /** * Shared utility methods for Iceberg maintenance jobs. * - *

Provides SQL escaping, argument parsing, and Spark configuration utilities used by both {@link - * IcebergRewriteDataFilesJob} and {@link IcebergExpireSnapshotsJob}. + *

Provides SQL escaping, argument parsing, and Spark configuration utilities used by {@link + * IcebergRewriteDataFilesJob}, {@link IcebergExpireSnapshotsJob} and {@link + * IcebergRewriteManifestsJob}. */ public final class IcebergJobUtils { + /** + * Matches a job template placeholder that no job configuration value replaced, e.g. {@code + * {{use_caching}}}. + */ + private static final Pattern UNRESOLVED_PLACEHOLDER_PATTERN = Pattern.compile("^\\{\\{[^{}]*}}$"); + private IcebergJobUtils() {} + /** + * Return the given argument value unless it is an unresolved job template placeholder. + * + *

Job templates declare optional parameters as {@code {{name}}} placeholders. When the caller + * omits a parameter, the server leaves the placeholder untouched and it reaches the job as a + * literal {@code "{{name}}"} argument. Treating that as a real value would forward nonsense to + * Iceberg, so callers use this method to map it back to "not supplied". + * + * @param value the argument value to inspect + * @return the value, or null if it is null or an unresolved placeholder + */ + public static String nullIfUnresolvedPlaceholder(String value) { + if (value == null || UNRESOLVED_PLACEHOLDER_PATTERN.matcher(value.trim()).matches()) { + return null; + } + return value; + } + /** * Escape single quotes in SQL string literals by replacing ' with ''. * diff --git a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteManifestsJob.java b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteManifestsJob.java index 0caa662fc5a..2cd677301b6 100644 --- a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteManifestsJob.java +++ b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteManifestsJob.java @@ -20,7 +20,6 @@ import java.util.Arrays; import java.util.Collections; -import java.util.HashMap; import java.util.List; import java.util.Map; import org.apache.gravitino.job.JobTemplateProvider; @@ -33,8 +32,8 @@ /** * Built-in job for rewriting Iceberg table manifest files. * - *

This job leverages Iceberg's RewriteManifestsProcedure to consolidate small manifest files - * and rewrite manifests with improved partition specs, which improves scan planning performance. + *

This job leverages Iceberg's RewriteManifestsProcedure to consolidate small manifest files and + * rewrite manifests with improved partition specs, which improves scan planning performance. */ public class IcebergRewriteManifestsJob implements BuiltInJob { @@ -65,17 +64,28 @@ public SparkJobTemplate jobTemplate() { *

    *
  • --catalog <catalog_name> Required. Iceberg catalog name. *
  • --table <table_identifier> Required. Table name (db.table) - *
  • --use-caching <boolean> Optional. Whether to use caching (default: true) + *
  • --use-caching <boolean> Optional. Whether to cache the table metadata in Spark + * while rewriting (default: Iceberg's own default) *
  • --spark-conf <spark_conf_json> Optional. JSON map of custom Spark configurations *
* + *

Important Notes on Special Characters: + * + *

    + *
  • Via Gravitino API: Pass values as-is without shell escaping. Gravitino handles + * escaping internally via ProcessBuilder. + *
  • Via Command Line: Use shell quoting for values containing whitespace. + *
+ * + *

Example via command line: --catalog iceberg_catalog --table db.sample --use-caching false + * *

Example via Gravitino API: * * 

{@code
    * Map jobConf = new HashMap<>();
    * jobConf.put("catalog_name", "iceberg_catalog");
    * jobConf.put("table_identifier", "db.sample");
-   * jobConf.put("use_caching", "true");
+   * jobConf.put("use_caching", "false");
    * metalake.runJob("builtin-iceberg-rewrite-manifests", jobConf);
    * }
*/ @@ -85,8 +95,10 @@ public static void main(String[] args) { System.exit(1); } - Map argMap = parseArguments(args); + // Parse named arguments + Map argMap = IcebergJobUtils.parseArguments(args); + // Validate required arguments String catalogName = argMap.get("catalog"); String tableIdentifier = argMap.get("table"); @@ -96,16 +108,28 @@ public static void main(String[] args) { System.exit(1); } - String useCaching = argMap.get("use-caching"); - String sparkConfJson = argMap.get("spark-conf"); + // Optional arguments. Unresolved template placeholders mean the caller left the parameter out, + // so they are dropped rather than forwarded to Iceberg as literal values. + String useCaching = IcebergJobUtils.nullIfUnresolvedPlaceholder(argMap.get("use-caching")); + String sparkConfJson = IcebergJobUtils.nullIfUnresolvedPlaceholder(argMap.get("spark-conf")); + // Validate use-caching if provided + try { + validateUseCaching(useCaching); + } catch (IllegalArgumentException e) { + System.err.println("Error: " + e.getMessage()); + printUsage(); + System.exit(1); + } + + // Build Spark session with custom configs if provided SparkSession.Builder sparkBuilder = SparkSession.builder().appName("Gravitino Built-in Iceberg Rewrite Manifests"); + // Apply custom Spark configurations if provided if (sparkConfJson != null && !sparkConfJson.isEmpty()) { try { - Map customConfigs = - IcebergRewriteDataFilesJob.parseCustomSparkConfigs(sparkConfJson); + Map customConfigs = IcebergJobUtils.parseCustomSparkConfigs(sparkConfJson); for (Map.Entry entry : customConfigs.entrySet()) { sparkBuilder.config(entry.getKey(), entry.getValue()); } @@ -120,18 +144,23 @@ public static void main(String[] args) { SparkSession spark = sparkBuilder.getOrCreate(); try { + // Build the procedure call SQL String sql = buildProcedureCall(catalogName, tableIdentifier, useCaching); + System.out.println("Executing Iceberg rewrite_manifests procedure: " + sql); - Row[] results = (Row[]) spark.sql(sql).collect(); + // Execute the procedure + List results = spark.sql(sql).collectAsList(); - if (results.length > 0) { - Row result = results[0]; + // Print results. The procedure output columns are numeric, but their exact width is an + // Iceberg implementation detail, so read them as Number rather than a fixed primitive. + if (!results.isEmpty()) { + Row result = results.get(0); System.out.printf( "Rewrite Manifests Results:%n" + " Rewritten manifests: %d%n" + " Added manifests: %d%n", - result.getInt(0), result.getInt(1)); + ((Number) result.get(0)).longValue(), ((Number) result.get(1)).longValue()); } System.out.println("Rewrite manifests job completed successfully"); @@ -149,18 +178,15 @@ public static void main(String[] args) { * * @param catalogName Iceberg catalog name * @param tableIdentifier Fully qualified table name - * @param useCaching Whether to use caching during rewrite + * @param useCaching Whether to cache table metadata during the rewrite * @return SQL CALL statement */ - static String buildProcedureCall( - String catalogName, String tableIdentifier, String useCaching) { + static String buildProcedureCall(String catalogName, String tableIdentifier, String useCaching) { StringBuilder sql = new StringBuilder(); sql.append("CALL ") - .append(IcebergRewriteDataFilesJob.escapeSqlIdentifier(catalogName)) + .append(IcebergJobUtils.escapeSqlIdentifier(catalogName)) .append(".system.rewrite_manifests("); - sql.append("table => '") - .append(IcebergRewriteDataFilesJob.escapeSqlString(tableIdentifier)) - .append("'"); + sql.append("table => '").append(IcebergJobUtils.escapeSqlString(tableIdentifier)).append("'"); if (useCaching != null && !useCaching.isEmpty()) { sql.append(", use_caching => ").append(Boolean.parseBoolean(useCaching)); @@ -171,30 +197,26 @@ static String buildProcedureCall( } /** - * Parse command line arguments in --key value format. + * Validate the use-caching parameter value. * - * @param args command line arguments - * @return map of argument names to values + *

{@link Boolean#parseBoolean(String)} maps anything that is not {@code "true"} to {@code + * false}, so a typo would silently disable caching. Reject such values instead. + * + * @param useCaching the use-caching value to validate + * @throws IllegalArgumentException if the value is neither "true" nor "false" */ - static Map parseArguments(String[] args) { - Map argMap = new HashMap<>(); - for (int i = 0; i < args.length; i++) { - if (args[i].startsWith("--")) { - String key = args[i].substring(2); - if (i + 1 < args.length && !args[i + 1].startsWith("--")) { - String value = args[i + 1]; - if (value != null && !value.trim().isEmpty()) { - argMap.put(key, value); - } - i++; - } else { - System.err.println("Warning: Flag " + args[i] + " has no value, ignoring"); - } - } + static void validateUseCaching(String useCaching) { + if (useCaching == null || useCaching.isEmpty()) { + return; // use-caching is optional + } + + if (!"true".equalsIgnoreCase(useCaching) && !"false".equalsIgnoreCase(useCaching)) { + throw new IllegalArgumentException( + "Invalid use-caching value '" + useCaching + "'. Must be either 'true' or 'false'"); } - return argMap; } + /** Print usage information. */ private static void printUsage() { System.err.println( "Usage: IcebergRewriteManifestsJob [OPTIONS]\n" @@ -204,18 +226,26 @@ private static void printUsage() { + " --table Fully qualified table name (e.g., db.table_name)\n" + "\n" + "Optional Options:\n" - + " --use-caching Whether to use caching during rewrite (default: true)\n" + + " --use-caching Cache table metadata in Spark while rewriting\n" + + " Must be either 'true' or 'false'\n" + + " Default: true (Iceberg default)\n" + " --spark-conf JSON map of custom Spark configurations\n" + " Example: '{\"spark.sql.shuffle.partitions\":\"200\"}'\n" + + " Note: Overriding required catalog/extensions/app-name configs is unsupported\n" + "\n" + "Examples:\n" - + " # Basic rewrite\n" + + " # Basic rewrite with Iceberg defaults\n" + " --catalog iceberg_prod --table db.sample\n" + "\n" - + " # Without caching\n" - + " --catalog iceberg_prod --table db.sample --use-caching false\n"); + + " # Rewrite without caching table metadata\n" + + " --catalog iceberg_prod --table db.sample --use-caching false"); } + /** + * Build template arguments list with named argument format. + * + * @return list of template arguments + */ private static List buildArguments() { return Arrays.asList( "--catalog", @@ -228,7 +258,12 @@ private static List buildArguments() { "{{spark_conf}}"); } + /** + * Build Spark configuration template. + * + * @return map of Spark configuration keys to template values + */ private static Map buildSparkConfigs() { return IcebergSparkConfigUtils.buildTemplateSparkConfigs(); } -} \ No newline at end of file +} diff --git a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteManifestsJob.java b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteManifestsJob.java index 30e19357ac9..9e2a43bbb48 100644 --- a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteManifestsJob.java +++ b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteManifestsJob.java @@ -18,9 +18,12 @@ */ 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; @@ -60,7 +63,7 @@ public void testJobTemplateHasExecutable() { } @Test - public void testJobTemplateHasMainClass() { + public void testJobTemplateHasClassName() { IcebergRewriteManifestsJob job = new IcebergRewriteManifestsJob(); SparkJobTemplate template = job.jobTemplate(); @@ -138,7 +141,7 @@ public void testJobTemplateNameMatchesBuiltInPattern() { @Test public void testParseArgumentsWithAllRequired() { String[] args = {"--catalog", "iceberg_prod", "--table", "db.sample"}; - Map result = IcebergRewriteManifestsJob.parseArguments(args); + Map result = IcebergJobUtils.parseArguments(args); assertEquals(2, result.size()); assertEquals("iceberg_prod", result.get("catalog")); @@ -152,7 +155,7 @@ public void testParseArgumentsWithOptional() { "--table", "db.sample", "--use-caching", "false" }; - Map result = IcebergRewriteManifestsJob.parseArguments(args); + Map result = IcebergJobUtils.parseArguments(args); assertEquals(3, result.size()); assertEquals("iceberg_prod", result.get("catalog")); @@ -163,7 +166,7 @@ public void testParseArgumentsWithOptional() { @Test public void testParseArgumentsWithEmptyValues() { String[] args = {"--catalog", "iceberg_prod", "--table", "db.sample", "--use-caching", ""}; - Map result = IcebergRewriteManifestsJob.parseArguments(args); + Map result = IcebergJobUtils.parseArguments(args); // Empty values should be ignored assertEquals(2, result.size()); @@ -172,45 +175,96 @@ public void testParseArgumentsWithEmptyValues() { assertFalse(result.containsKey("use-caching")); } - @Test - public void testParseArgumentsWithMissingValues() { - String[] args = {"--catalog", "iceberg_prod", "--table"}; - Map result = IcebergRewriteManifestsJob.parseArguments(args); - - // Only catalog should be parsed, table has no value - assertEquals(1, result.size()); - assertEquals("iceberg_prod", result.get("catalog")); - assertFalse(result.containsKey("table")); - } - @Test public void testParseArgumentsOrderIndependent() { String[] args1 = {"--catalog", "cat1", "--table", "tbl1", "--use-caching", "true"}; String[] args2 = {"--use-caching", "true", "--table", "tbl1", "--catalog", "cat1"}; - Map result1 = IcebergRewriteManifestsJob.parseArguments(args1); - Map result2 = IcebergRewriteManifestsJob.parseArguments(args2); + Map result1 = IcebergJobUtils.parseArguments(args1); + Map result2 = IcebergJobUtils.parseArguments(args2); assertEquals(result1, result2); } + // Test unresolved placeholder handling + + @Test + public void testUnresolvedPlaceholderIsDroppedForOptionalArguments() { + // The server leaves placeholders untouched when the caller omits the parameter, so they reach + // the job as literal arguments. + String[] args = { + "--catalog", "iceberg_prod", + "--table", "db.sample", + "--use-caching", "{{use_caching}}", + "--spark-conf", "{{spark_conf}}" + }; + Map result = IcebergJobUtils.parseArguments(args); + + assertEquals("{{use_caching}}", result.get("use-caching")); + assertNull(IcebergJobUtils.nullIfUnresolvedPlaceholder(result.get("use-caching"))); + assertNull(IcebergJobUtils.nullIfUnresolvedPlaceholder(result.get("spark-conf"))); + } + + @Test + public void testRealValuesSurviveUnresolvedPlaceholderFiltering() { + assertEquals("false", IcebergJobUtils.nullIfUnresolvedPlaceholder("false")); + assertEquals("db.sample", IcebergJobUtils.nullIfUnresolvedPlaceholder("db.sample")); + // Only a value that is entirely a placeholder is dropped. + assertEquals("{{a}}b", IcebergJobUtils.nullIfUnresolvedPlaceholder("{{a}}b")); + assertNull(IcebergJobUtils.nullIfUnresolvedPlaceholder(null)); + } + + @Test + public void testProcedureCallOmitsUseCachingWhenPlaceholderUnresolved() { + String useCaching = IcebergJobUtils.nullIfUnresolvedPlaceholder("{{use_caching}}"); + String sql = + IcebergRewriteManifestsJob.buildProcedureCall("iceberg_prod", "db.sample", useCaching); + + // Without filtering, Boolean.parseBoolean("{{use_caching}}") would silently emit + // use_caching => false instead of leaving Iceberg's default in place. + assertEquals("CALL `iceberg_prod`.system.rewrite_manifests(table => 'db.sample')", sql); + } + + // Test validateUseCaching method + + @Test + public void testValidateUseCachingAcceptsBooleans() { + assertDoesNotThrow(() -> IcebergRewriteManifestsJob.validateUseCaching("true")); + assertDoesNotThrow(() -> IcebergRewriteManifestsJob.validateUseCaching("false")); + assertDoesNotThrow(() -> IcebergRewriteManifestsJob.validateUseCaching("TRUE")); + } + + @Test + public void testValidateUseCachingAcceptsAbsentValue() { + assertDoesNotThrow(() -> IcebergRewriteManifestsJob.validateUseCaching(null)); + assertDoesNotThrow(() -> IcebergRewriteManifestsJob.validateUseCaching("")); + } + + @Test + public void testValidateUseCachingRejectsNonBoolean() { + IllegalArgumentException e = + assertThrows( + IllegalArgumentException.class, + () -> IcebergRewriteManifestsJob.validateUseCaching("yes")); + + assertTrue(e.getMessage().contains("yes")); + } + // Test buildProcedureCall method @Test public void testBuildProcedureCallMinimal() { - String sql = - IcebergRewriteManifestsJob.buildProcedureCall("iceberg_prod", "db.sample", null); + String sql = IcebergRewriteManifestsJob.buildProcedureCall("iceberg_prod", "db.sample", null); - assertEquals("CALL iceberg_prod.system.rewrite_manifests(table => 'db.sample')", sql); + assertEquals("CALL `iceberg_prod`.system.rewrite_manifests(table => 'db.sample')", sql); } @Test public void testBuildProcedureCallWithCachingTrue() { - String sql = - IcebergRewriteManifestsJob.buildProcedureCall("iceberg_prod", "db.sample", "true"); + String sql = IcebergRewriteManifestsJob.buildProcedureCall("iceberg_prod", "db.sample", "true"); assertEquals( - "CALL iceberg_prod.system.rewrite_manifests(table => 'db.sample', use_caching => true)", + "CALL `iceberg_prod`.system.rewrite_manifests(table => 'db.sample', use_caching => true)", sql); } @@ -220,16 +274,15 @@ public void testBuildProcedureCallWithCachingFalse() { IcebergRewriteManifestsJob.buildProcedureCall("iceberg_prod", "db.sample", "false"); assertEquals( - "CALL iceberg_prod.system.rewrite_manifests(table => 'db.sample', use_caching => false)", + "CALL `iceberg_prod`.system.rewrite_manifests(table => 'db.sample', use_caching => false)", sql); } @Test public void testBuildProcedureCallWithEmptyCaching() { - String sql = - IcebergRewriteManifestsJob.buildProcedureCall("iceberg_prod", "db.sample", ""); + String sql = IcebergRewriteManifestsJob.buildProcedureCall("iceberg_prod", "db.sample", ""); - assertEquals("CALL iceberg_prod.system.rewrite_manifests(table => 'db.sample')", sql); + assertEquals("CALL `iceberg_prod`.system.rewrite_manifests(table => 'db.sample')", sql); } @Test @@ -245,10 +298,9 @@ public void testBuildProcedureCallWithSqlInjectionAttempt() { // Test SQL injection attempt in catalog name String maliciousCatalog = "catalog`; DROP TABLE users; --"; - sql = - IcebergRewriteManifestsJob.buildProcedureCall(maliciousCatalog, "db.table", null); + sql = IcebergRewriteManifestsJob.buildProcedureCall(maliciousCatalog, "db.table", null); - // Verify backticks are escaped - assertTrue(sql.contains("catalog``; DROP TABLE users; --")); + // Verify backticks are escaped and the whole identifier stays quoted + assertTrue(sql.startsWith("CALL `catalog``; DROP TABLE users; --`.system.rewrite_manifests(")); } -} \ No newline at end of file +} From e7261d21b980eb8286bb22368a49bb129a5c7523 Mon Sep 17 00:00:00 2001 From: Akshay Thorat Date: Sun, 6 Sep 2026 19:54:12 -0700 Subject: [PATCH 4/7] feat(maintenance): support spec_id in rewrite-manifests job MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Iceberg's rewrite_manifests procedure takes three in-parameters: table, use_caching, and spec_id. Only the first two were reachable, which left the "align manifests with a better partition spec" half of #11196 unavailable - a spec evolution had no way to re-cluster old manifests onto the new spec. Expose it as jobConf key spec_id, validated as a non-negative integer so a malformed value fails before the Spark session is created rather than reaching Spark as an unparseable SQL literal. Like use_caching, an unresolved {{spec_id}} placeholder is dropped so an omitted parameter leaves the table's current spec in force. Co-authored-by: İbrahim ERBİLEN --- .../optimizer-cli-reference.md | 8 +- .../iceberg/IcebergRewriteManifestsJob.java | 51 ++++++++++- .../TestIcebergRewriteManifestsJob.java | 86 +++++++++++++++++-- 3 files changed, 131 insertions(+), 14 deletions(-) diff --git a/docs/table-maintenance-service/optimizer-cli-reference.md b/docs/table-maintenance-service/optimizer-cli-reference.md index 25603990ea8..7709d9b8a8d 100644 --- a/docs/table-maintenance-service/optimizer-cli-reference.md +++ b/docs/table-maintenance-service/optimizer-cli-reference.md @@ -365,10 +365,13 @@ The job calls Iceberg's `rewrite_manifests` stored procedure through Spark SQL. | `catalog_name` | Iceberg catalog name as registered in Spark | Required | | `table_identifier` | Fully qualified table name, such as `db.sample` | Required | | `use_caching` | Caches table metadata in Spark while rewriting; `true` or `false` | `true` (Iceberg default) | +| `spec_id` | Partition spec ID to rewrite manifests to; non-negative integer | The table's current spec | | `spark_conf` | JSON map of Spark configuration | None | Leave `use_caching` unset unless the driver is memory constrained. Caching keeps the table metadata in Spark for the duration of the rewrite, which is faster but holds more memory. +Leave `spec_id` unset for routine consolidation. Set it after a partition spec evolution, when manifests written under the old spec should be re-clustered onto a specific one; `GET .../tables/{table}` and the table's `partitions` metadata table show which spec IDs exist. An unknown ID fails the job rather than falling back to the current spec. + ### Submitting the Job ```bash @@ -392,12 +395,13 @@ curl -X POST -H "Accept: application/vnd.gravitino.v1+json" \ http://localhost:8090/api/metalakes/test/jobs ``` -The job builds this statement, including `use_caching` only when you supply it: +The job builds this statement, including `use_caching` and `spec_id` only when you supply them: ```sql CALL `rest_catalog`.system.rewrite_manifests( table => 'db.t1', - use_caching => false + use_caching => false, + spec_id => 2 ) ``` diff --git a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteManifestsJob.java b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteManifestsJob.java index 2cd677301b6..4f4e92f9d88 100644 --- a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteManifestsJob.java +++ b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteManifestsJob.java @@ -66,6 +66,8 @@ public SparkJobTemplate jobTemplate() { *

  • --table <table_identifier> Required. Table name (db.table) *
  • --use-caching <boolean> Optional. Whether to cache the table metadata in Spark * while rewriting (default: Iceberg's own default) + *
  • --spec-id <int> Optional. Rewrite manifests to this partition spec ID (default: the + * table's current spec) *
  • --spark-conf <spark_conf_json> Optional. JSON map of custom Spark configurations * * @@ -111,11 +113,13 @@ public static void main(String[] args) { // Optional arguments. Unresolved template placeholders mean the caller left the parameter out, // so they are dropped rather than forwarded to Iceberg as literal values. String useCaching = IcebergJobUtils.nullIfUnresolvedPlaceholder(argMap.get("use-caching")); + String specId = IcebergJobUtils.nullIfUnresolvedPlaceholder(argMap.get("spec-id")); String sparkConfJson = IcebergJobUtils.nullIfUnresolvedPlaceholder(argMap.get("spark-conf")); - // Validate use-caching if provided + // Validate optional arguments if provided try { validateUseCaching(useCaching); + validateSpecId(specId); } catch (IllegalArgumentException e) { System.err.println("Error: " + e.getMessage()); printUsage(); @@ -145,7 +149,7 @@ public static void main(String[] args) { try { // Build the procedure call SQL - String sql = buildProcedureCall(catalogName, tableIdentifier, useCaching); + String sql = buildProcedureCall(catalogName, tableIdentifier, useCaching, specId); System.out.println("Executing Iceberg rewrite_manifests procedure: " + sql); @@ -179,9 +183,11 @@ public static void main(String[] args) { * @param catalogName Iceberg catalog name * @param tableIdentifier Fully qualified table name * @param useCaching Whether to cache table metadata during the rewrite + * @param specId Partition spec ID to rewrite manifests to * @return SQL CALL statement */ - static String buildProcedureCall(String catalogName, String tableIdentifier, String useCaching) { + static String buildProcedureCall( + String catalogName, String tableIdentifier, String useCaching, String specId) { StringBuilder sql = new StringBuilder(); sql.append("CALL ") .append(IcebergJobUtils.escapeSqlIdentifier(catalogName)) @@ -192,6 +198,10 @@ static String buildProcedureCall(String catalogName, String tableIdentifier, Str sql.append(", use_caching => ").append(Boolean.parseBoolean(useCaching)); } + if (specId != null && !specId.isEmpty()) { + sql.append(", spec_id => ").append(Integer.parseInt(specId)); + } + sql.append(")"); return sql.toString(); } @@ -216,6 +226,31 @@ static void validateUseCaching(String useCaching) { } } + /** + * Validate the spec-id parameter value. + * + *

    Iceberg partition spec IDs start at 0 and the procedure rejects an unknown ID, but failing + * here keeps a malformed value from reaching Spark as an unparseable SQL literal. + * + * @param specId the spec-id value to validate + * @throws IllegalArgumentException if the value is not a non-negative integer + */ + static void validateSpecId(String specId) { + if (specId == null || specId.isEmpty()) { + return; // spec-id is optional + } + + try { + if (Integer.parseInt(specId) < 0) { + throw new IllegalArgumentException( + "Invalid spec-id value '" + specId + "'. Must be a non-negative integer"); + } + } catch (NumberFormatException e) { + throw new IllegalArgumentException( + "Invalid spec-id value '" + specId + "'. Must be a non-negative integer"); + } + } + /** Print usage information. */ private static void printUsage() { System.err.println( @@ -229,6 +264,9 @@ private static void printUsage() { + " --use-caching Cache table metadata in Spark while rewriting\n" + " Must be either 'true' or 'false'\n" + " Default: true (Iceberg default)\n" + + " --spec-id Rewrite manifests to this partition spec ID\n" + + " Must be a non-negative integer\n" + + " Default: the table's current partition spec\n" + " --spark-conf JSON map of custom Spark configurations\n" + " Example: '{\"spark.sql.shuffle.partitions\":\"200\"}'\n" + " Note: Overriding required catalog/extensions/app-name configs is unsupported\n" @@ -238,7 +276,10 @@ private static void printUsage() { + " --catalog iceberg_prod --table db.sample\n" + "\n" + " # Rewrite without caching table metadata\n" - + " --catalog iceberg_prod --table db.sample --use-caching false"); + + " --catalog iceberg_prod --table db.sample --use-caching false\n" + + "\n" + + " # Re-cluster manifests onto partition spec 2 after a spec evolution\n" + + " --catalog iceberg_prod --table db.sample --spec-id 2"); } /** @@ -254,6 +295,8 @@ private static List buildArguments() { "{{table_identifier}}", "--use-caching", "{{use_caching}}", + "--spec-id", + "{{spec_id}}", "--spark-conf", "{{spark_conf}}"); } diff --git a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteManifestsJob.java b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteManifestsJob.java index 9e2a43bbb48..baea1e05e80 100644 --- a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteManifestsJob.java +++ b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteManifestsJob.java @@ -77,7 +77,7 @@ public void testJobTemplateHasArguments() { SparkJobTemplate template = job.jobTemplate(); assertNotNull(template.arguments()); - assertEquals(8, template.arguments().size()); // 4 flags * 2 (flag + value) + assertEquals(10, template.arguments().size()); // 5 flags * 2 (flag + value) // Verify all expected arguments are present assertTrue(template.arguments().contains("--catalog")); @@ -86,6 +86,8 @@ public void testJobTemplateHasArguments() { 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}}")); } @@ -218,7 +220,8 @@ public void testRealValuesSurviveUnresolvedPlaceholderFiltering() { public void testProcedureCallOmitsUseCachingWhenPlaceholderUnresolved() { String useCaching = IcebergJobUtils.nullIfUnresolvedPlaceholder("{{use_caching}}"); String sql = - IcebergRewriteManifestsJob.buildProcedureCall("iceberg_prod", "db.sample", useCaching); + IcebergRewriteManifestsJob.buildProcedureCall( + "iceberg_prod", "db.sample", useCaching, null); // Without filtering, Boolean.parseBoolean("{{use_caching}}") would silently emit // use_caching => false instead of leaving Iceberg's default in place. @@ -250,18 +253,53 @@ public void testValidateUseCachingRejectsNonBoolean() { assertTrue(e.getMessage().contains("yes")); } + // Test validateSpecId method + + @Test + public void testValidateSpecIdAcceptsNonNegativeIntegers() { + assertDoesNotThrow(() -> IcebergRewriteManifestsJob.validateSpecId("0")); + assertDoesNotThrow(() -> IcebergRewriteManifestsJob.validateSpecId("2")); + } + + @Test + public void testValidateSpecIdAcceptsAbsentValue() { + assertDoesNotThrow(() -> IcebergRewriteManifestsJob.validateSpecId(null)); + assertDoesNotThrow(() -> IcebergRewriteManifestsJob.validateSpecId("")); + } + + @Test + public void testValidateSpecIdRejectsNegativeValue() { + IllegalArgumentException e = + assertThrows( + IllegalArgumentException.class, () -> IcebergRewriteManifestsJob.validateSpecId("-1")); + + assertTrue(e.getMessage().contains("-1")); + } + + @Test + public void testValidateSpecIdRejectsNonNumericValue() { + IllegalArgumentException e = + assertThrows( + IllegalArgumentException.class, + () -> IcebergRewriteManifestsJob.validateSpecId("latest")); + + assertTrue(e.getMessage().contains("latest")); + } + // Test buildProcedureCall method @Test public void testBuildProcedureCallMinimal() { - String sql = IcebergRewriteManifestsJob.buildProcedureCall("iceberg_prod", "db.sample", null); + String sql = + IcebergRewriteManifestsJob.buildProcedureCall("iceberg_prod", "db.sample", null, null); assertEquals("CALL `iceberg_prod`.system.rewrite_manifests(table => 'db.sample')", sql); } @Test public void testBuildProcedureCallWithCachingTrue() { - String sql = IcebergRewriteManifestsJob.buildProcedureCall("iceberg_prod", "db.sample", "true"); + String sql = + IcebergRewriteManifestsJob.buildProcedureCall("iceberg_prod", "db.sample", "true", null); assertEquals( "CALL `iceberg_prod`.system.rewrite_manifests(table => 'db.sample', use_caching => true)", @@ -271,7 +309,7 @@ public void testBuildProcedureCallWithCachingTrue() { @Test public void testBuildProcedureCallWithCachingFalse() { String sql = - IcebergRewriteManifestsJob.buildProcedureCall("iceberg_prod", "db.sample", "false"); + IcebergRewriteManifestsJob.buildProcedureCall("iceberg_prod", "db.sample", "false", null); assertEquals( "CALL `iceberg_prod`.system.rewrite_manifests(table => 'db.sample', use_caching => false)", @@ -280,8 +318,39 @@ public void testBuildProcedureCallWithCachingFalse() { @Test public void testBuildProcedureCallWithEmptyCaching() { - String sql = IcebergRewriteManifestsJob.buildProcedureCall("iceberg_prod", "db.sample", ""); + String sql = + IcebergRewriteManifestsJob.buildProcedureCall("iceberg_prod", "db.sample", "", null); + + assertEquals("CALL `iceberg_prod`.system.rewrite_manifests(table => 'db.sample')", sql); + } + + @Test + public void testBuildProcedureCallWithSpecId() { + String sql = + IcebergRewriteManifestsJob.buildProcedureCall("iceberg_prod", "db.sample", null, "2"); + + assertEquals( + "CALL `iceberg_prod`.system.rewrite_manifests(table => 'db.sample', spec_id => 2)", sql); + } + + @Test + public void testBuildProcedureCallWithCachingAndSpecId() { + String sql = + IcebergRewriteManifestsJob.buildProcedureCall("iceberg_prod", "db.sample", "false", "0"); + + assertEquals( + "CALL `iceberg_prod`.system.rewrite_manifests(table => 'db.sample', " + + "use_caching => false, spec_id => 0)", + sql); + } + + @Test + public void testProcedureCallOmitsSpecIdWhenPlaceholderUnresolved() { + String specId = IcebergJobUtils.nullIfUnresolvedPlaceholder("{{spec_id}}"); + String sql = + IcebergRewriteManifestsJob.buildProcedureCall("iceberg_prod", "db.sample", null, specId); + // An unresolved placeholder must not reach Integer.parseInt as a SQL literal. assertEquals("CALL `iceberg_prod`.system.rewrite_manifests(table => 'db.sample')", sql); } @@ -290,7 +359,8 @@ public void testBuildProcedureCallWithSqlInjectionAttempt() { // Test SQL injection attempt in table name String maliciousTable = "db.table' OR '1'='1"; String sql = - IcebergRewriteManifestsJob.buildProcedureCall("iceberg_catalog", maliciousTable, null); + IcebergRewriteManifestsJob.buildProcedureCall( + "iceberg_catalog", maliciousTable, null, null); // Verify single quotes are escaped (becomes '') assertTrue(sql.contains("db.table'' OR ''1''=''1")); @@ -298,7 +368,7 @@ public void testBuildProcedureCallWithSqlInjectionAttempt() { // Test SQL injection attempt in catalog name String maliciousCatalog = "catalog`; DROP TABLE users; --"; - sql = IcebergRewriteManifestsJob.buildProcedureCall(maliciousCatalog, "db.table", null); + sql = IcebergRewriteManifestsJob.buildProcedureCall(maliciousCatalog, "db.table", null, null); // Verify backticks are escaped and the whole identifier stays quoted assertTrue(sql.startsWith("CALL `catalog``; DROP TABLE users; --`.system.rewrite_manifests(")); From 3d6cf0985561fc9997701aef4fb859fb918144b2 Mon Sep 17 00:00:00 2001 From: Akshay Thorat Date: Mon, 14 Sep 2026 10:48:38 -0700 Subject: [PATCH 5/7] docs: clarify manifest spec selection and runtime defaults MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Document spec discovery and correct CLI examples and help text without changing execution behavior. Co-authored-by: İbrahim ERBİLEN --- .../optimizer-cli-reference.md | 56 ++++++++++++------- .../iceberg/IcebergRewriteManifestsJob.java | 14 ++--- 2 files changed, 43 insertions(+), 27 deletions(-) diff --git a/docs/table-maintenance-service/optimizer-cli-reference.md b/docs/table-maintenance-service/optimizer-cli-reference.md index 7709d9b8a8d..119417949e3 100644 --- a/docs/table-maintenance-service/optimizer-cli-reference.md +++ b/docs/table-maintenance-service/optimizer-cli-reference.md @@ -343,34 +343,49 @@ Expire Snapshots Results: ## Rewrite Manifests -`builtin-iceberg-rewrite-manifests` consolidates a table's manifest files. Frequent commits leave behind many small manifests, and scan planning has to open every one of them to decide which data files a filter matches. Rewriting them into fewer, larger manifests aligned with the current partition spec cuts that planning cost. +`builtin-iceberg-rewrite-manifests` consolidates a table's manifest files. Frequent writes can leave many small manifests. Scan planning uses manifest-list summaries to prune them, then reads the remaining manifests. Rewriting consolidates and clusters entries for one existing partition spec without repartitioning data files. This complements `builtin-iceberg-rewrite-data-files`: that job improves the data file layout, this one improves the metadata that points at it. The job calls Iceberg's `rewrite_manifests` stored procedure through Spark SQL. -| Property | Value | -|-------------|------------------------------------------------------------------------------| -| Name | `builtin-iceberg-rewrite-manifests` | -| Type | Spark | -| Version | `v1` | -| Main class | `org.apache.gravitino.maintenance.jobs.iceberg.IcebergRewriteManifestsJob` | +| Property | Value | +| ------------- | ------------------------------------------------------------------------------ | +| Name | `builtin-iceberg-rewrite-manifests` | +| Type | Spark | +| Version | `v1` | +| Main class | `org.apache.gravitino.maintenance.jobs.iceberg.IcebergRewriteManifestsJob` | ### Parameters `catalog_name` and `table_identifier` are required. The rest are optional. -| Key | Description | Default | -|--------------------|--------------------------------------------------------------------|-----------------------------| -| `catalog_name` | Iceberg catalog name as registered in Spark | Required | -| `table_identifier` | Fully qualified table name, such as `db.sample` | Required | -| `use_caching` | Caches table metadata in Spark while rewriting; `true` or `false` | `true` (Iceberg default) | -| `spec_id` | Partition spec ID to rewrite manifests to; non-negative integer | The table's current spec | -| `spark_conf` | JSON map of Spark configuration | None | +| Key | Description | Default | +| -------------------- | ------------------------------------------------------------------------ | ----------------------------------- | +| `catalog_name` | Iceberg catalog name as registered in Spark | Required | +| `table_identifier` | Fully qualified table name, such as `db.sample` | Required | +| `use_caching` | Caches table metadata in Spark while rewriting; `true` or `false` | Installed Iceberg version's default | +| `spec_id` | Existing partition spec whose manifests to rewrite; non-negative integer | The table's current spec | +| `spark_conf` | JSON map of Spark configuration | None | + +Leave `use_caching` unset to use the installed Iceberg version's default (`false` in Iceberg 1.11.0). Set it explicitly when consistent behavior across versions is required. + +### How to find `spec_id` + +Omit `spec_id` for routine maintenance of the current partition spec. Do not guess IDs such as `0` or `1`. Read `default-spec-id` and the `partition-specs` array from the table's current Iceberg metadata JSON. The array maps each `spec-id` to its partition fields and transforms; `default-spec-id` identifies the current spec. These are Iceberg table metadata fields, not Gravitino catalog properties. + +To discover which specs have manifests in the current snapshot, run: + +```sql +SELECT DISTINCT partition_spec_id +FROM rest_catalog.db.t1.manifests; +``` + +This query lists represented specs, not which one is current. A defined spec with no manifests may be absent; use the metadata JSON to identify the default and interpret the transforms. -Leave `use_caching` unset unless the driver is memory constrained. Caching keeps the table metadata in Spark for the duration of the rewrite, which is faster but holds more memory. +For example, suppose metadata shows spec `0` uses `day(event_time)` and the current spec `1` uses `hour(event_time)`. Omitting `spec_id` rewrites eligible manifests for spec `1`. Passing `"spec_id": "0"` consolidates the old day-spec manifests. Neither run converts day-partitioned data files to hour partitioning or rewrites manifests belonging to the other spec. -Leave `spec_id` unset for routine consolidation. Set it after a partition spec evolution, when manifests written under the old spec should be re-clustered onto a specific one; `GET .../tables/{table}` and the table's `partitions` metadata table show which spec IDs exist. An unknown ID fails the job rather than falling back to the current spec. +`spec_id` selects the existing spec whose manifests are eligible for rewriting, and replacement manifests use that same spec. Iceberg validates that the ID exists. This follows the [Iceberg 1.11.0 action implementation](https://github.com/apache/iceberg/blob/apache-iceberg-1.11.0/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteManifestsSparkAction.java), where `findMatchingManifests` compares each manifest's `partitionSpecId()` to the selected spec. It is not a partition-evolution operation. ### Submitting the Job @@ -392,7 +407,7 @@ curl -X POST -H "Accept: application/vnd.gravitino.v1+json" \ "warehouse_location": "" } }' \ - http://localhost:8090/api/metalakes/test/jobs + http://localhost:8090/api/metalakes/test/jobs/runs ``` The job builds this statement, including `use_caching` and `spec_id` only when you supply them: @@ -408,8 +423,8 @@ CALL `rest_catalog`.system.rewrite_manifests( ### Verifying the Result ```bash -curl -sS "http://localhost:8090/api/metalakes/test/jobs/{job_id}" | jq '.job.state' -cat /tmp/gravitino/jobs/staging/test/builtin-iceberg-rewrite-manifests/{job_id}/stdout.log +curl -sS "http://localhost:8090/api/metalakes/test/jobs/runs/{job_id}" | jq '.job.status' +cat /tmp/gravitino/jobs/staging/test/builtin-iceberg-rewrite-manifests/{job_id}/output.log ``` A successful run reports its state as `SUCCEEDED` and logs how many manifests it replaced: @@ -420,7 +435,8 @@ Rewrite Manifests Results: Added manifests: 2 ``` -Both counts at zero means the table's manifests were already consolidated, which is a successful no-op rather than a failure. +Both counts at zero indicate a successful no-op, for example when no manifests for the selected spec need rewriting. Other specs remain unchanged. + ## Related diff --git a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteManifestsJob.java b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteManifestsJob.java index 4f4e92f9d88..fe5cbaeb072 100644 --- a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteManifestsJob.java +++ b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteManifestsJob.java @@ -33,7 +33,7 @@ * Built-in job for rewriting Iceberg table manifest files. * *

    This job leverages Iceberg's RewriteManifestsProcedure to consolidate small manifest files and - * rewrite manifests with improved partition specs, which improves scan planning performance. + * cluster manifest entries within an existing partition spec to improve scan planning. */ public class IcebergRewriteManifestsJob implements BuiltInJob { @@ -66,8 +66,8 @@ public SparkJobTemplate jobTemplate() { *

  • --table <table_identifier> Required. Table name (db.table) *
  • --use-caching <boolean> Optional. Whether to cache the table metadata in Spark * while rewriting (default: Iceberg's own default) - *
  • --spec-id <int> Optional. Rewrite manifests to this partition spec ID (default: the - * table's current spec) + *
  • --spec-id <int> Optional. Rewrite manifests belonging to this partition spec ID + * (default: the table's current spec) *
  • --spark-conf <spark_conf_json> Optional. JSON map of custom Spark configurations * * @@ -183,7 +183,7 @@ public static void main(String[] args) { * @param catalogName Iceberg catalog name * @param tableIdentifier Fully qualified table name * @param useCaching Whether to cache table metadata during the rewrite - * @param specId Partition spec ID to rewrite manifests to + * @param specId Existing partition spec ID whose manifests to rewrite * @return SQL CALL statement */ static String buildProcedureCall( @@ -263,8 +263,8 @@ private static void printUsage() { + "Optional Options:\n" + " --use-caching Cache table metadata in Spark while rewriting\n" + " Must be either 'true' or 'false'\n" - + " Default: true (Iceberg default)\n" - + " --spec-id Rewrite manifests to this partition spec ID\n" + + " Default: installed Iceberg version's default\n" + + " --spec-id Rewrite manifests belonging to this partition spec ID\n" + " Must be a non-negative integer\n" + " Default: the table's current partition spec\n" + " --spark-conf JSON map of custom Spark configurations\n" @@ -278,7 +278,7 @@ private static void printUsage() { + " # Rewrite without caching table metadata\n" + " --catalog iceberg_prod --table db.sample --use-caching false\n" + "\n" - + " # Re-cluster manifests onto partition spec 2 after a spec evolution\n" + + " # Consolidate manifests belonging to existing partition spec 2\n" + " --catalog iceberg_prod --table db.sample --spec-id 2"); } From 481b6152d1600be29c044d3e13878686c418e8bd Mon Sep 17 00:00:00 2001 From: Akshay Thorat Date: Wed, 16 Sep 2026 10:17:01 -0700 Subject: [PATCH 6/7] fix(maintenance): align manifest job keys with CLI flags MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Use matching argument placeholders and the catalog key in Spark configuration while preserving defaults for existing jobs. Co-authored-by: İbrahim ERBİLEN --- .../optimizer-cli-reference.md | 28 +++++------ .../iceberg/IcebergRewriteManifestsJob.java | 18 +++---- .../TestIcebergRewriteManifestsJob.java | 47 +++++++++++++------ .../common/util/IcebergSparkConfigUtils.java | 20 ++++++-- 4 files changed, 71 insertions(+), 42 deletions(-) diff --git a/docs/table-maintenance-service/optimizer-cli-reference.md b/docs/table-maintenance-service/optimizer-cli-reference.md index 119417949e3..3c7a5a6cb12 100644 --- a/docs/table-maintenance-service/optimizer-cli-reference.md +++ b/docs/table-maintenance-service/optimizer-cli-reference.md @@ -358,21 +358,21 @@ The job calls Iceberg's `rewrite_manifests` stored procedure through Spark SQL. ### Parameters -`catalog_name` and `table_identifier` are required. The rest are optional. +`catalog` and `table` are required. The rest are optional. | Key | Description | Default | | -------------------- | ------------------------------------------------------------------------ | ----------------------------------- | -| `catalog_name` | Iceberg catalog name as registered in Spark | Required | -| `table_identifier` | Fully qualified table name, such as `db.sample` | Required | -| `use_caching` | Caches table metadata in Spark while rewriting; `true` or `false` | Installed Iceberg version's default | -| `spec_id` | Existing partition spec whose manifests to rewrite; non-negative integer | The table's current spec | -| `spark_conf` | JSON map of Spark configuration | None | +| `catalog` | Iceberg catalog name as registered in Spark | Required | +| `table` | Fully qualified table name, such as `db.sample` | Required | +| `use-caching` | Caches table metadata in Spark while rewriting; `true` or `false` | Installed Iceberg version's default | +| `spec-id` | Existing partition spec whose manifests to rewrite; non-negative integer | The table's current spec | +| `spark-conf` | JSON map of Spark configuration | None | -Leave `use_caching` unset to use the installed Iceberg version's default (`false` in Iceberg 1.11.0). Set it explicitly when consistent behavior across versions is required. +Leave `use-caching` unset to use the installed Iceberg version's default (`false` in Iceberg 1.11.0). Set it explicitly when consistent behavior across versions is required. -### How to find `spec_id` +### How to find `spec-id` -Omit `spec_id` for routine maintenance of the current partition spec. Do not guess IDs such as `0` or `1`. Read `default-spec-id` and the `partition-specs` array from the table's current Iceberg metadata JSON. The array maps each `spec-id` to its partition fields and transforms; `default-spec-id` identifies the current spec. These are Iceberg table metadata fields, not Gravitino catalog properties. +Omit `spec-id` for routine maintenance of the current partition spec. Do not guess IDs such as `0` or `1`. Read `default-spec-id` and the `partition-specs` array from the table's current Iceberg metadata JSON. The array maps each `spec-id` to its partition fields and transforms; `default-spec-id` identifies the current spec. These are Iceberg table metadata fields, not Gravitino catalog properties. To discover which specs have manifests in the current snapshot, run: @@ -383,9 +383,9 @@ FROM rest_catalog.db.t1.manifests; This query lists represented specs, not which one is current. A defined spec with no manifests may be absent; use the metadata JSON to identify the default and interpret the transforms. -For example, suppose metadata shows spec `0` uses `day(event_time)` and the current spec `1` uses `hour(event_time)`. Omitting `spec_id` rewrites eligible manifests for spec `1`. Passing `"spec_id": "0"` consolidates the old day-spec manifests. Neither run converts day-partitioned data files to hour partitioning or rewrites manifests belonging to the other spec. +For example, suppose metadata shows spec `0` uses `day(event_time)` and the current spec `1` uses `hour(event_time)`. Omitting `spec-id` rewrites eligible manifests for spec `1`. Passing `"spec-id": "0"` consolidates the old day-spec manifests. Neither run converts day-partitioned data files to hour partitioning or rewrites manifests belonging to the other spec. -`spec_id` selects the existing spec whose manifests are eligible for rewriting, and replacement manifests use that same spec. Iceberg validates that the ID exists. This follows the [Iceberg 1.11.0 action implementation](https://github.com/apache/iceberg/blob/apache-iceberg-1.11.0/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteManifestsSparkAction.java), where `findMatchingManifests` compares each manifest's `partitionSpecId()` to the selected spec. It is not a partition-evolution operation. +`spec-id` selects the existing spec whose manifests are eligible for rewriting, and replacement manifests use that same spec. Iceberg validates that the ID exists. This follows the [Iceberg 1.11.0 action implementation](https://github.com/apache/iceberg/blob/apache-iceberg-1.11.0/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteManifestsSparkAction.java), where `findMatchingManifests` compares each manifest's `partitionSpecId()` to the selected spec. It is not a partition-evolution operation. ### Submitting the Job @@ -395,8 +395,8 @@ curl -X POST -H "Accept: application/vnd.gravitino.v1+json" \ -d '{ "jobTemplateName": "builtin-iceberg-rewrite-manifests", "jobConf": { - "catalog_name": "rest_catalog", - "table_identifier": "db.t1", + "catalog": "rest_catalog", + "table": "db.t1", "spark_master": "local[2]", "spark_executor_instances": "1", "spark_executor_cores": "1", @@ -410,7 +410,7 @@ curl -X POST -H "Accept: application/vnd.gravitino.v1+json" \ http://localhost:8090/api/metalakes/test/jobs/runs ``` -The job builds this statement, including `use_caching` and `spec_id` only when you supply them: +The job builds this statement, including `use-caching` and `spec-id` only when you supply them: ```sql CALL `rest_catalog`.system.rewrite_manifests( diff --git a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteManifestsJob.java b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteManifestsJob.java index fe5cbaeb072..4c9f926d96e 100644 --- a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteManifestsJob.java +++ b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteManifestsJob.java @@ -85,9 +85,9 @@ public SparkJobTemplate jobTemplate() { * * 
    {@code
    * Map jobConf = new HashMap<>();
-   * jobConf.put("catalog_name", "iceberg_catalog");
-   * jobConf.put("table_identifier", "db.sample");
-   * jobConf.put("use_caching", "false");
+   * jobConf.put("catalog", "iceberg_catalog");
+   * jobConf.put("table", "db.sample");
+   * jobConf.put("use-caching", "false");
    * metalake.runJob("builtin-iceberg-rewrite-manifests", jobConf);
    * }
    */ @@ -290,15 +290,15 @@ private static void printUsage() { private static List buildArguments() { return Arrays.asList( "--catalog", - "{{catalog_name}}", + "{{catalog}}", "--table", - "{{table_identifier}}", + "{{table}}", "--use-caching", - "{{use_caching}}", + "{{use-caching}}", "--spec-id", - "{{spec_id}}", + "{{spec-id}}", "--spark-conf", - "{{spark_conf}}"); + "{{spark-conf}}"); } /** @@ -307,6 +307,6 @@ private static List buildArguments() { * @return map of Spark configuration keys to template values */ private static Map buildSparkConfigs() { - return IcebergSparkConfigUtils.buildTemplateSparkConfigs(); + return IcebergSparkConfigUtils.buildTemplateSparkConfigs("catalog"); } } diff --git a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteManifestsJob.java b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteManifestsJob.java index baea1e05e80..ba640f087da 100644 --- a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteManifestsJob.java +++ b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteManifestsJob.java @@ -29,10 +29,27 @@ import java.util.Map; import org.apache.gravitino.job.JobTemplateProvider; import org.apache.gravitino.job.SparkJobTemplate; +import org.apache.gravitino.maintenance.optimizer.common.util.IcebergSparkConfigUtils; import org.junit.jupiter.api.Test; public class TestIcebergRewriteManifestsJob { + @Test + public void testArgumentKeysMatchFlagsAndCatalogConfig() { + SparkJobTemplate template = new IcebergRewriteManifestsJob().jobTemplate(); + for (int i = 0; i < template.arguments().size(); i += 2) { + String key = template.arguments().get(i).substring(2); + assertEquals("{{" + key + "}}", template.arguments().get(i + 1)); + } + assertEquals( + "org.apache.iceberg.spark.SparkCatalog", + template.configs().get("spark.sql.catalog.{{catalog}}")); + assertFalse(template.configs().keySet().stream().anyMatch(key -> key.contains("catalog_name"))); + assertTrue( + IcebergSparkConfigUtils.buildTemplateSparkConfigs() + .containsKey("spark.sql.catalog.{{catalog_name}}")); + } + @Test public void testJobTemplateHasCorrectName() { IcebergRewriteManifestsJob job = new IcebergRewriteManifestsJob(); @@ -81,15 +98,15 @@ public void testJobTemplateHasArguments() { // Verify all expected arguments are present assertTrue(template.arguments().contains("--catalog")); - assertTrue(template.arguments().contains("{{catalog_name}}")); + assertTrue(template.arguments().contains("{{catalog}}")); assertTrue(template.arguments().contains("--table")); - assertTrue(template.arguments().contains("{{table_identifier}}")); + assertTrue(template.arguments().contains("{{table}}")); assertTrue(template.arguments().contains("--use-caching")); - 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("{{spec-id}}")); assertTrue(template.arguments().contains("--spark-conf")); - assertTrue(template.arguments().contains("{{spark_conf}}")); + assertTrue(template.arguments().contains("{{spark-conf}}")); } @Test @@ -109,10 +126,10 @@ public void testJobTemplateHasSparkConfigs() { 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")); + assertTrue(configs.containsKey("spark.sql.catalog.{{catalog}}")); + assertTrue(configs.containsKey("spark.sql.catalog.{{catalog}}.type")); + assertTrue(configs.containsKey("spark.sql.catalog.{{catalog}}.uri")); + assertTrue(configs.containsKey("spark.sql.catalog.{{catalog}}.warehouse")); } @Test @@ -197,12 +214,12 @@ public void testUnresolvedPlaceholderIsDroppedForOptionalArguments() { String[] args = { "--catalog", "iceberg_prod", "--table", "db.sample", - "--use-caching", "{{use_caching}}", - "--spark-conf", "{{spark_conf}}" + "--use-caching", "{{use-caching}}", + "--spark-conf", "{{spark-conf}}" }; Map result = IcebergJobUtils.parseArguments(args); - assertEquals("{{use_caching}}", result.get("use-caching")); + assertEquals("{{use-caching}}", result.get("use-caching")); assertNull(IcebergJobUtils.nullIfUnresolvedPlaceholder(result.get("use-caching"))); assertNull(IcebergJobUtils.nullIfUnresolvedPlaceholder(result.get("spark-conf"))); } @@ -218,12 +235,12 @@ public void testRealValuesSurviveUnresolvedPlaceholderFiltering() { @Test public void testProcedureCallOmitsUseCachingWhenPlaceholderUnresolved() { - String useCaching = IcebergJobUtils.nullIfUnresolvedPlaceholder("{{use_caching}}"); + String useCaching = IcebergJobUtils.nullIfUnresolvedPlaceholder("{{use-caching}}"); String sql = IcebergRewriteManifestsJob.buildProcedureCall( "iceberg_prod", "db.sample", useCaching, null); - // Without filtering, Boolean.parseBoolean("{{use_caching}}") would silently emit + // Without filtering, Boolean.parseBoolean("{{use-caching}}") would silently emit // use_caching => false instead of leaving Iceberg's default in place. assertEquals("CALL `iceberg_prod`.system.rewrite_manifests(table => 'db.sample')", sql); } @@ -346,7 +363,7 @@ public void testBuildProcedureCallWithCachingAndSpecId() { @Test public void testProcedureCallOmitsSpecIdWhenPlaceholderUnresolved() { - String specId = IcebergJobUtils.nullIfUnresolvedPlaceholder("{{spec_id}}"); + String specId = IcebergJobUtils.nullIfUnresolvedPlaceholder("{{spec-id}}"); String sql = IcebergRewriteManifestsJob.buildProcedureCall("iceberg_prod", "db.sample", null, specId); diff --git a/maintenance/optimizer-api/src/main/java/org/apache/gravitino/maintenance/optimizer/common/util/IcebergSparkConfigUtils.java b/maintenance/optimizer-api/src/main/java/org/apache/gravitino/maintenance/optimizer/common/util/IcebergSparkConfigUtils.java index 0e03e3ecc3a..2850dbf6fed 100644 --- a/maintenance/optimizer-api/src/main/java/org/apache/gravitino/maintenance/optimizer/common/util/IcebergSparkConfigUtils.java +++ b/maintenance/optimizer-api/src/main/java/org/apache/gravitino/maintenance/optimizer/common/util/IcebergSparkConfigUtils.java @@ -49,6 +49,17 @@ private IcebergSparkConfigUtils() {} /** Build default Spark template configs for Iceberg jobs. */ public static Map buildTemplateSparkConfigs() { + return buildTemplateSparkConfigs("catalog_name"); + } + + /** + * Build Spark template configs using the given catalog job configuration key. + * + * @param catalogKey job configuration key identifying the Spark catalog + * @return immutable Spark configuration template + */ + public static Map buildTemplateSparkConfigs(String catalogKey) { + String catalogPlaceholder = "{{" + catalogKey + "}}"; Map configs = new HashMap<>(); configs.put("spark.master", "{{spark_master}}"); configs.put("spark.executor.instances", "{{spark_executor_instances}}"); @@ -56,10 +67,11 @@ public static Map buildTemplateSparkConfigs() { configs.put("spark.executor.memory", "{{spark_executor_memory}}"); configs.put("spark.driver.memory", "{{spark_driver_memory}}"); configs.put( - SPARK_SQL_CATALOG_PREFIX + "{{catalog_name}}", "org.apache.iceberg.spark.SparkCatalog"); - configs.put(SPARK_SQL_CATALOG_PREFIX + "{{catalog_name}}.type", "{{catalog_type}}"); - configs.put(SPARK_SQL_CATALOG_PREFIX + "{{catalog_name}}.uri", "{{catalog_uri}}"); - configs.put(SPARK_SQL_CATALOG_PREFIX + "{{catalog_name}}.warehouse", "{{warehouse_location}}"); + SPARK_SQL_CATALOG_PREFIX + catalogPlaceholder, "org.apache.iceberg.spark.SparkCatalog"); + configs.put(SPARK_SQL_CATALOG_PREFIX + catalogPlaceholder + ".type", "{{catalog_type}}"); + configs.put(SPARK_SQL_CATALOG_PREFIX + catalogPlaceholder + ".uri", "{{catalog_uri}}"); + configs.put( + SPARK_SQL_CATALOG_PREFIX + catalogPlaceholder + ".warehouse", "{{warehouse_location}}"); configs.put(SPARK_SQL_EXTENSIONS_KEY, ICEBERG_SPARK_EXTENSIONS); return Collections.unmodifiableMap(configs); } From c4ea3f0fd6fed5b2ed982c85c30404a2d6e9df1e Mon Sep 17 00:00:00 2001 From: Akshay Thorat Date: Thu, 17 Sep 2026 11:03:57 -0700 Subject: [PATCH 7/7] fix(maintenance): restore original manifest job configuration keys MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Revert CLI/key alignment after the review restriction was withdrawn. Restore the shared catalog helper and original tests and examples. Co-authored-by: İbrahim ERBİLEN --- .../optimizer-cli-reference.md | 28 +++++------ .../iceberg/IcebergRewriteManifestsJob.java | 18 +++---- .../TestIcebergRewriteManifestsJob.java | 47 ++++++------------- .../common/util/IcebergSparkConfigUtils.java | 20 ++------ 4 files changed, 42 insertions(+), 71 deletions(-) diff --git a/docs/table-maintenance-service/optimizer-cli-reference.md b/docs/table-maintenance-service/optimizer-cli-reference.md index 3c7a5a6cb12..119417949e3 100644 --- a/docs/table-maintenance-service/optimizer-cli-reference.md +++ b/docs/table-maintenance-service/optimizer-cli-reference.md @@ -358,21 +358,21 @@ The job calls Iceberg's `rewrite_manifests` stored procedure through Spark SQL. ### Parameters -`catalog` and `table` are required. The rest are optional. +`catalog_name` and `table_identifier` are required. The rest are optional. | Key | Description | Default | | -------------------- | ------------------------------------------------------------------------ | ----------------------------------- | -| `catalog` | Iceberg catalog name as registered in Spark | Required | -| `table` | Fully qualified table name, such as `db.sample` | Required | -| `use-caching` | Caches table metadata in Spark while rewriting; `true` or `false` | Installed Iceberg version's default | -| `spec-id` | Existing partition spec whose manifests to rewrite; non-negative integer | The table's current spec | -| `spark-conf` | JSON map of Spark configuration | None | +| `catalog_name` | Iceberg catalog name as registered in Spark | Required | +| `table_identifier` | Fully qualified table name, such as `db.sample` | Required | +| `use_caching` | Caches table metadata in Spark while rewriting; `true` or `false` | Installed Iceberg version's default | +| `spec_id` | Existing partition spec whose manifests to rewrite; non-negative integer | The table's current spec | +| `spark_conf` | JSON map of Spark configuration | None | -Leave `use-caching` unset to use the installed Iceberg version's default (`false` in Iceberg 1.11.0). Set it explicitly when consistent behavior across versions is required. +Leave `use_caching` unset to use the installed Iceberg version's default (`false` in Iceberg 1.11.0). Set it explicitly when consistent behavior across versions is required. -### How to find `spec-id` +### How to find `spec_id` -Omit `spec-id` for routine maintenance of the current partition spec. Do not guess IDs such as `0` or `1`. Read `default-spec-id` and the `partition-specs` array from the table's current Iceberg metadata JSON. The array maps each `spec-id` to its partition fields and transforms; `default-spec-id` identifies the current spec. These are Iceberg table metadata fields, not Gravitino catalog properties. +Omit `spec_id` for routine maintenance of the current partition spec. Do not guess IDs such as `0` or `1`. Read `default-spec-id` and the `partition-specs` array from the table's current Iceberg metadata JSON. The array maps each `spec-id` to its partition fields and transforms; `default-spec-id` identifies the current spec. These are Iceberg table metadata fields, not Gravitino catalog properties. To discover which specs have manifests in the current snapshot, run: @@ -383,9 +383,9 @@ FROM rest_catalog.db.t1.manifests; This query lists represented specs, not which one is current. A defined spec with no manifests may be absent; use the metadata JSON to identify the default and interpret the transforms. -For example, suppose metadata shows spec `0` uses `day(event_time)` and the current spec `1` uses `hour(event_time)`. Omitting `spec-id` rewrites eligible manifests for spec `1`. Passing `"spec-id": "0"` consolidates the old day-spec manifests. Neither run converts day-partitioned data files to hour partitioning or rewrites manifests belonging to the other spec. +For example, suppose metadata shows spec `0` uses `day(event_time)` and the current spec `1` uses `hour(event_time)`. Omitting `spec_id` rewrites eligible manifests for spec `1`. Passing `"spec_id": "0"` consolidates the old day-spec manifests. Neither run converts day-partitioned data files to hour partitioning or rewrites manifests belonging to the other spec. -`spec-id` selects the existing spec whose manifests are eligible for rewriting, and replacement manifests use that same spec. Iceberg validates that the ID exists. This follows the [Iceberg 1.11.0 action implementation](https://github.com/apache/iceberg/blob/apache-iceberg-1.11.0/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteManifestsSparkAction.java), where `findMatchingManifests` compares each manifest's `partitionSpecId()` to the selected spec. It is not a partition-evolution operation. +`spec_id` selects the existing spec whose manifests are eligible for rewriting, and replacement manifests use that same spec. Iceberg validates that the ID exists. This follows the [Iceberg 1.11.0 action implementation](https://github.com/apache/iceberg/blob/apache-iceberg-1.11.0/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteManifestsSparkAction.java), where `findMatchingManifests` compares each manifest's `partitionSpecId()` to the selected spec. It is not a partition-evolution operation. ### Submitting the Job @@ -395,8 +395,8 @@ curl -X POST -H "Accept: application/vnd.gravitino.v1+json" \ -d '{ "jobTemplateName": "builtin-iceberg-rewrite-manifests", "jobConf": { - "catalog": "rest_catalog", - "table": "db.t1", + "catalog_name": "rest_catalog", + "table_identifier": "db.t1", "spark_master": "local[2]", "spark_executor_instances": "1", "spark_executor_cores": "1", @@ -410,7 +410,7 @@ curl -X POST -H "Accept: application/vnd.gravitino.v1+json" \ http://localhost:8090/api/metalakes/test/jobs/runs ``` -The job builds this statement, including `use-caching` and `spec-id` only when you supply them: +The job builds this statement, including `use_caching` and `spec_id` only when you supply them: ```sql CALL `rest_catalog`.system.rewrite_manifests( diff --git a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteManifestsJob.java b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteManifestsJob.java index 4c9f926d96e..fe5cbaeb072 100644 --- a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteManifestsJob.java +++ b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteManifestsJob.java @@ -85,9 +85,9 @@ public SparkJobTemplate jobTemplate() { * * 
    {@code
    * Map jobConf = new HashMap<>();
-   * jobConf.put("catalog", "iceberg_catalog");
-   * jobConf.put("table", "db.sample");
-   * jobConf.put("use-caching", "false");
+   * jobConf.put("catalog_name", "iceberg_catalog");
+   * jobConf.put("table_identifier", "db.sample");
+   * jobConf.put("use_caching", "false");
    * metalake.runJob("builtin-iceberg-rewrite-manifests", jobConf);
    * }
    */ @@ -290,15 +290,15 @@ private static void printUsage() { private static List buildArguments() { return Arrays.asList( "--catalog", - "{{catalog}}", + "{{catalog_name}}", "--table", - "{{table}}", + "{{table_identifier}}", "--use-caching", - "{{use-caching}}", + "{{use_caching}}", "--spec-id", - "{{spec-id}}", + "{{spec_id}}", "--spark-conf", - "{{spark-conf}}"); + "{{spark_conf}}"); } /** @@ -307,6 +307,6 @@ private static List buildArguments() { * @return map of Spark configuration keys to template values */ private static Map buildSparkConfigs() { - return IcebergSparkConfigUtils.buildTemplateSparkConfigs("catalog"); + return IcebergSparkConfigUtils.buildTemplateSparkConfigs(); } } diff --git a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteManifestsJob.java b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteManifestsJob.java index ba640f087da..baea1e05e80 100644 --- a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteManifestsJob.java +++ b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteManifestsJob.java @@ -29,27 +29,10 @@ import java.util.Map; import org.apache.gravitino.job.JobTemplateProvider; import org.apache.gravitino.job.SparkJobTemplate; -import org.apache.gravitino.maintenance.optimizer.common.util.IcebergSparkConfigUtils; import org.junit.jupiter.api.Test; public class TestIcebergRewriteManifestsJob { - @Test - public void testArgumentKeysMatchFlagsAndCatalogConfig() { - SparkJobTemplate template = new IcebergRewriteManifestsJob().jobTemplate(); - for (int i = 0; i < template.arguments().size(); i += 2) { - String key = template.arguments().get(i).substring(2); - assertEquals("{{" + key + "}}", template.arguments().get(i + 1)); - } - assertEquals( - "org.apache.iceberg.spark.SparkCatalog", - template.configs().get("spark.sql.catalog.{{catalog}}")); - assertFalse(template.configs().keySet().stream().anyMatch(key -> key.contains("catalog_name"))); - assertTrue( - IcebergSparkConfigUtils.buildTemplateSparkConfigs() - .containsKey("spark.sql.catalog.{{catalog_name}}")); - } - @Test public void testJobTemplateHasCorrectName() { IcebergRewriteManifestsJob job = new IcebergRewriteManifestsJob(); @@ -98,15 +81,15 @@ public void testJobTemplateHasArguments() { // Verify all expected arguments are present assertTrue(template.arguments().contains("--catalog")); - assertTrue(template.arguments().contains("{{catalog}}")); + assertTrue(template.arguments().contains("{{catalog_name}}")); assertTrue(template.arguments().contains("--table")); - 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("{{use_caching}}")); assertTrue(template.arguments().contains("--spec-id")); - assertTrue(template.arguments().contains("{{spec-id}}")); + assertTrue(template.arguments().contains("{{spec_id}}")); assertTrue(template.arguments().contains("--spark-conf")); - assertTrue(template.arguments().contains("{{spark-conf}}")); + assertTrue(template.arguments().contains("{{spark_conf}}")); } @Test @@ -126,10 +109,10 @@ public void testJobTemplateHasSparkConfigs() { assertTrue(configs.containsKey("spark.driver.memory")); // Verify Iceberg catalog configs - assertTrue(configs.containsKey("spark.sql.catalog.{{catalog}}")); - assertTrue(configs.containsKey("spark.sql.catalog.{{catalog}}.type")); - assertTrue(configs.containsKey("spark.sql.catalog.{{catalog}}.uri")); - assertTrue(configs.containsKey("spark.sql.catalog.{{catalog}}.warehouse")); + 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 @@ -214,12 +197,12 @@ public void testUnresolvedPlaceholderIsDroppedForOptionalArguments() { String[] args = { "--catalog", "iceberg_prod", "--table", "db.sample", - "--use-caching", "{{use-caching}}", - "--spark-conf", "{{spark-conf}}" + "--use-caching", "{{use_caching}}", + "--spark-conf", "{{spark_conf}}" }; Map result = IcebergJobUtils.parseArguments(args); - assertEquals("{{use-caching}}", result.get("use-caching")); + assertEquals("{{use_caching}}", result.get("use-caching")); assertNull(IcebergJobUtils.nullIfUnresolvedPlaceholder(result.get("use-caching"))); assertNull(IcebergJobUtils.nullIfUnresolvedPlaceholder(result.get("spark-conf"))); } @@ -235,12 +218,12 @@ public void testRealValuesSurviveUnresolvedPlaceholderFiltering() { @Test public void testProcedureCallOmitsUseCachingWhenPlaceholderUnresolved() { - String useCaching = IcebergJobUtils.nullIfUnresolvedPlaceholder("{{use-caching}}"); + String useCaching = IcebergJobUtils.nullIfUnresolvedPlaceholder("{{use_caching}}"); String sql = IcebergRewriteManifestsJob.buildProcedureCall( "iceberg_prod", "db.sample", useCaching, null); - // Without filtering, Boolean.parseBoolean("{{use-caching}}") would silently emit + // Without filtering, Boolean.parseBoolean("{{use_caching}}") would silently emit // use_caching => false instead of leaving Iceberg's default in place. assertEquals("CALL `iceberg_prod`.system.rewrite_manifests(table => 'db.sample')", sql); } @@ -363,7 +346,7 @@ public void testBuildProcedureCallWithCachingAndSpecId() { @Test public void testProcedureCallOmitsSpecIdWhenPlaceholderUnresolved() { - String specId = IcebergJobUtils.nullIfUnresolvedPlaceholder("{{spec-id}}"); + String specId = IcebergJobUtils.nullIfUnresolvedPlaceholder("{{spec_id}}"); String sql = IcebergRewriteManifestsJob.buildProcedureCall("iceberg_prod", "db.sample", null, specId); diff --git a/maintenance/optimizer-api/src/main/java/org/apache/gravitino/maintenance/optimizer/common/util/IcebergSparkConfigUtils.java b/maintenance/optimizer-api/src/main/java/org/apache/gravitino/maintenance/optimizer/common/util/IcebergSparkConfigUtils.java index 2850dbf6fed..0e03e3ecc3a 100644 --- a/maintenance/optimizer-api/src/main/java/org/apache/gravitino/maintenance/optimizer/common/util/IcebergSparkConfigUtils.java +++ b/maintenance/optimizer-api/src/main/java/org/apache/gravitino/maintenance/optimizer/common/util/IcebergSparkConfigUtils.java @@ -49,17 +49,6 @@ private IcebergSparkConfigUtils() {} /** Build default Spark template configs for Iceberg jobs. */ public static Map buildTemplateSparkConfigs() { - return buildTemplateSparkConfigs("catalog_name"); - } - - /** - * Build Spark template configs using the given catalog job configuration key. - * - * @param catalogKey job configuration key identifying the Spark catalog - * @return immutable Spark configuration template - */ - public static Map buildTemplateSparkConfigs(String catalogKey) { - String catalogPlaceholder = "{{" + catalogKey + "}}"; Map configs = new HashMap<>(); configs.put("spark.master", "{{spark_master}}"); configs.put("spark.executor.instances", "{{spark_executor_instances}}"); @@ -67,11 +56,10 @@ public static Map buildTemplateSparkConfigs(String catalogKey) { configs.put("spark.executor.memory", "{{spark_executor_memory}}"); configs.put("spark.driver.memory", "{{spark_driver_memory}}"); configs.put( - SPARK_SQL_CATALOG_PREFIX + catalogPlaceholder, "org.apache.iceberg.spark.SparkCatalog"); - configs.put(SPARK_SQL_CATALOG_PREFIX + catalogPlaceholder + ".type", "{{catalog_type}}"); - configs.put(SPARK_SQL_CATALOG_PREFIX + catalogPlaceholder + ".uri", "{{catalog_uri}}"); - configs.put( - SPARK_SQL_CATALOG_PREFIX + catalogPlaceholder + ".warehouse", "{{warehouse_location}}"); + SPARK_SQL_CATALOG_PREFIX + "{{catalog_name}}", "org.apache.iceberg.spark.SparkCatalog"); + configs.put(SPARK_SQL_CATALOG_PREFIX + "{{catalog_name}}.type", "{{catalog_type}}"); + configs.put(SPARK_SQL_CATALOG_PREFIX + "{{catalog_name}}.uri", "{{catalog_uri}}"); + configs.put(SPARK_SQL_CATALOG_PREFIX + "{{catalog_name}}.warehouse", "{{warehouse_location}}"); configs.put(SPARK_SQL_EXTENSIONS_KEY, ICEBERG_SPARK_EXTENSIONS); return Collections.unmodifiableMap(configs); }