Copilot commented on code in PR #11216:
URL: https://github.com/apache/gravitino/pull/11216#discussion_r3299436034
##########
maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/BuiltInJobTemplateProvider.java:
##########
@@ -30,6 +30,7 @@
import org.apache.gravitino.maintenance.jobs.spark.SparkPiJob;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import
org.apache.gravitino.maintenance.jobs.iceberg.IcebergRewriteManifestsJob;
Review Comment:
The new import is placed after the `org.slf4j` imports, breaking the
alphabetical/grouped import ordering used elsewhere in the file. Move it next
to the other `org.apache.gravitino.maintenance.jobs.iceberg.*` imports to keep
import ordering consistent.
##########
maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/BuiltInJobTemplateProvider.java:
##########
@@ -45,7 +46,8 @@ public class BuiltInJobTemplateProvider implements
JobTemplateProvider {
ImmutableList.of(
new SparkPiJob(),
new IcebergRewriteDataFilesJob(),
- new IcebergUpdateStatsAndMetricsJob());
+ new IcebergUpdateStatsAndMetricsJob()),
Review Comment:
The closing parenthesis of `ImmutableList.of(...)` is placed after
`IcebergUpdateStatsAndMetricsJob()`, so `new IcebergRewriteManifestsJob()` is
outside the list. This will not compile (extra trailing `)` and dangling
expression). Move the `)` to after `new IcebergRewriteManifestsJob()` so the
new job is actually included in the immutable list.
##########
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.
+ *
+ * <p>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.
+ *
+ * <p>Uses named arguments for flexibility:
+ *
+ * <ul>
+ * <li>--catalog <catalog_name> Required. Iceberg catalog name.
+ * <li>--table <table_identifier> Required. Table name (db.table)
+ * <li>--use-caching <boolean> Optional. Whether to use caching
(default: true)
+ * <li>--spark-conf <spark_conf_json> Optional. JSON map of custom
Spark configurations
+ * </ul>
+ *
+ * <p>Example via Gravitino API:
+ *
+ * <pre>{@code
+ * Map<String, String> jobConf = new HashMap<>();
+ * jobConf.put("catalog_name", "iceberg_catalog");
+ * jobConf.put("table_identifier", "db.sample");
+ * jobConf.put("use_caching", "true");
+ * metalake.runJob("builtin-iceberg-rewrite-manifests", jobConf);
+ * }</pre>
+ */
+ public static void main(String[] args) {
+ if (args.length < 4) {
+ printUsage();
+ System.exit(1);
+ }
+
+ Map<String, String> 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<String, String> customConfigs =
+ IcebergRewriteDataFilesJob.parseCustomSparkConfigs(sparkConfJson);
Review Comment:
This job reaches into `IcebergRewriteDataFilesJob` for
`parseCustomSparkConfigs`, `escapeSqlIdentifier`, and `escapeSqlString`.
Creating a cross-class dependency between two sibling job implementations
couples them and will make refactors error-prone. Consider extracting these
shared utilities into a dedicated helper class (e.g., `IcebergJobUtils`
alongside `IcebergSparkConfigUtils`) that both jobs depend on.
##########
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());
Review Comment:
The test method is named `testJobTemplateHasMainClass` but it asserts on
`template.className()`. Rename to `testJobTemplateHasClassName` to match the
property being verified.
##########
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.
+ *
+ * <p>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.
+ *
+ * <p>Uses named arguments for flexibility:
+ *
+ * <ul>
+ * <li>--catalog <catalog_name> Required. Iceberg catalog name.
+ * <li>--table <table_identifier> Required. Table name (db.table)
+ * <li>--use-caching <boolean> Optional. Whether to use caching
(default: true)
+ * <li>--spark-conf <spark_conf_json> Optional. JSON map of custom
Spark configurations
+ * </ul>
+ *
+ * <p>Example via Gravitino API:
+ *
+ * <pre>{@code
+ * Map<String, String> jobConf = new HashMap<>();
+ * jobConf.put("catalog_name", "iceberg_catalog");
+ * jobConf.put("table_identifier", "db.sample");
+ * jobConf.put("use_caching", "true");
+ * metalake.runJob("builtin-iceberg-rewrite-manifests", jobConf);
+ * }</pre>
+ */
+ public static void main(String[] args) {
+ if (args.length < 4) {
+ printUsage();
+ System.exit(1);
+ }
+
+ Map<String, String> 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<String, String> customConfigs =
+ IcebergRewriteDataFilesJob.parseCustomSparkConfigs(sparkConfJson);
+ for (Map.Entry<String, String> 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<String, String> parseArguments(String[] args) {
+ Map<String, String> 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++;
Review Comment:
When the value is empty/blank, the flag is silently skipped without a
warning, but `i` is still incremented (correct). However, the template
substitution `{{use_caching}}` and `{{spark_conf}}` may pass through literally
when not supplied by the user, resulting in these literal placeholder strings
being treated as valid values (they don't start with `--` and aren't empty).
Consider also filtering out unresolved `{{...}}` placeholders to avoid passing
them downstream (e.g., as `use_caching => {{use_caching}}` becoming a SQL parse
error). Optional but worth handling explicitly.
##########
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.
+ *
+ * <p>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.
+ *
+ * <p>Uses named arguments for flexibility:
+ *
+ * <ul>
+ * <li>--catalog <catalog_name> Required. Iceberg catalog name.
+ * <li>--table <table_identifier> Required. Table name (db.table)
+ * <li>--use-caching <boolean> Optional. Whether to use caching
(default: true)
+ * <li>--spark-conf <spark_conf_json> Optional. JSON map of custom
Spark configurations
+ * </ul>
+ *
+ * <p>Example via Gravitino API:
+ *
+ * <pre>{@code
+ * Map<String, String> jobConf = new HashMap<>();
+ * jobConf.put("catalog_name", "iceberg_catalog");
+ * jobConf.put("table_identifier", "db.sample");
+ * jobConf.put("use_caching", "true");
+ * metalake.runJob("builtin-iceberg-rewrite-manifests", jobConf);
+ * }</pre>
+ */
+ public static void main(String[] args) {
+ if (args.length < 4) {
+ printUsage();
+ System.exit(1);
+ }
+
+ Map<String, String> 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<String, String> customConfigs =
+ IcebergRewriteDataFilesJob.parseCustomSparkConfigs(sparkConfJson);
+ for (Map.Entry<String, String> 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));
Review Comment:
`Dataset<Row>.collect()` already returns `Row[]` in the Spark Java API, but
more importantly, if the procedure returns zero rows (e.g., nothing to rewrite)
the subsequent `result.getInt(0)`/`getInt(1)` calls would still be safe due to
the `length > 0` guard — however, the schema of `rewrite_manifests` returns
`rewritten_manifests_count` and `added_manifests_count` which may be
`long`/`int` depending on Iceberg version. Using `getInt` can throw
`ClassCastException` on some versions; consider using `getLong` or `get(0)`
with explicit conversion to be version-tolerant. The explicit `(Row[])` cast is
also redundant and can be removed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
