This is an automated email from the ASF dual-hosted git repository. github-actions[bot] pushed a commit to branch cherry-pick-35c184d2-to-branch-1.3 in repository https://gitbox.apache.org/repos/asf/gravitino.git
commit f1d4a68cd8fbd9497e554b1c9216e7c902216081 Author: MaSai <[email protected]> AuthorDate: Thu Sep 17 14:28:29 2026 +0800 [#13133] fix(maintenance): Fail Iceberg jobs when Spark runtime is missing (#13134) ### What changes were proposed in this pull request? Addresses #13133: 1. **Fail when Iceberg Spark runtime is missing** Built-in Iceberg jobs check for `IcebergSparkSessionExtensions` / `SparkCatalog` after `SparkSession` creation and exit non-zero when absent. 2. **Document the Iceberg Spark runtime prerequisite** Optimizer docs state that operators must supply a matching `iceberg-spark-runtime` (for example via `spark.jars` / Spark env), and troubleshooting reflects fail-fast behavior. 3. **Clearer optional-arg / JSON errors for callers** Iceberg JSON parse failures name the real CLI flags (`--updater-options`, `--spark-conf`, `--options`). Docs note that omitting optional `jobConf` keys can leave dangling `--flag` placeholders, so callers/UIs should send explicit values. ### Why are the changes needed? Without Iceberg Spark runtime, Spark only warns and jobs continue as if Iceberg were available. Docs did not make the runtime prerequisite discoverable. Optional template placeholders and opaque JSON parse errors also made failed submissions harder to diagnose. Fix: #13133 ### Does this PR introduce _any_ user-facing change? - Built-in Iceberg jobs now exit non-zero when Iceberg Spark runtime classes are missing. - Docs clarify the `iceberg-spark-runtime` prerequisite (`spark.jars` / Spark env), fail-fast behavior, and dangling optional template flags. - JSON option parse errors name the real `--flag` names. ### How was this patch tested? ```bash ./gradlew :maintenance:jobs:test \ --tests org.apache.gravitino.maintenance.jobs.iceberg.TestIcebergJobUtils \ --tests org.apache.gravitino.maintenance.jobs.iceberg.TestIcebergUpdateStatsJob \ --tests org.apache.gravitino.maintenance.jobs.iceberg.TestIcebergRewriteDataFilesJob \ --tests org.apache.gravitino.maintenance.jobs.iceberg.TestIcebergExpireSnapshotsJob \ -PskipITs -PskipDockerTests=true ``` --------- Co-authored-by: Cursor <[email protected]> # Conflicts: # maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergExpireSnapshotsJob.java # maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergJobUtils.java # maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergExpireSnapshotsJob.java --- .../optimizer-cli-reference.md | 16 + .../optimizer-configuration.md | 12 + .../optimizer-troubleshooting.md | 16 + docs/table-maintenance-service/optimizer.md | 11 + .../jobs/iceberg/IcebergExpireSnapshotsJob.java | 312 ++++++++++++++ .../maintenance/jobs/iceberg/IcebergJobUtils.java | 177 ++++++++ .../jobs/iceberg/IcebergRewriteDataFilesJob.java | 42 +- .../iceberg/IcebergUpdateStatsAndMetricsJob.java | 17 +- .../iceberg/TestIcebergExpireSnapshotsJob.java | 477 +++++++++++++++++++++ .../jobs/iceberg/TestIcebergJobUtils.java | 57 +++ .../iceberg/TestIcebergRewriteDataFilesJob.java | 4 +- .../jobs/iceberg/TestIcebergUpdateStatsJob.java | 43 +- 12 files changed, 1130 insertions(+), 54 deletions(-) diff --git a/docs/table-maintenance-service/optimizer-cli-reference.md b/docs/table-maintenance-service/optimizer-cli-reference.md index fcfa9130a0..2389cbb956 100644 --- a/docs/table-maintenance-service/optimizer-cli-reference.md +++ b/docs/table-maintenance-service/optimizer-cli-reference.md @@ -244,6 +244,22 @@ Three job templates ship with the service, and they are complementary rather tha 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. +These templates set Iceberg Spark session and catalog classes, but they do not list an Iceberg Spark +runtime in `jars`. `gravitino-jobs` also excludes that runtime from its shaded JAR, so the version +that runs with your Spark cluster is yours to supply. Provide a matching +`iceberg-spark-runtime-<sparkMajor>_<scala>` JAR on the Spark classpath used by the job executor — +commonly through `spark.jars` in `spark_conf`, or by installing it into `SPARK_HOME`. Align the +artifact with the Spark, Scala, and Iceberg versions you actually run. A reference coordinate used +in Gravitino's own jobs tests is `org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.11.0`. +Without that runtime, built-in Iceberg jobs fail after Spark starts instead of continuing without +Iceberg support. + +Optional template arguments are still listed as `--flag` + `{{placeholder}}` pairs. If `jobConf` +omits a key (or leaves the placeholder unresolved), the flag remains on the process command line as +a dangling argument (for example `--updater-options` with no value before `--spark-conf`). Callers +and UIs should supply every placeholder they care about with an explicit value, including optional +ones they intentionally disable or leave at a documented default, rather than omitting the key. + ## Update Statistics `builtin-iceberg-update-stats` reads a table and writes back the statistics and metrics that policies evaluate. Compaction policies read `custom-data-file-mse` and `custom-delete-file-number`, so nothing else will fire until this job has run at least once. diff --git a/docs/table-maintenance-service/optimizer-configuration.md b/docs/table-maintenance-service/optimizer-configuration.md index 409a0d70e5..cc379da68b 100644 --- a/docs/table-maintenance-service/optimizer-configuration.md +++ b/docs/table-maintenance-service/optimizer-configuration.md @@ -79,6 +79,18 @@ A direct job submission carries its own `jobConf`. This is `builtin-iceberg-upda `updater_options` and `spark_conf` are JSON strings inside a JSON object, so their quotes are escaped. That nesting is the most common source of malformed submissions. +Built-in Iceberg templates list optional keys as `--flag` + `{{placeholder}}`. Omitting a key from +`jobConf` does not remove that flag from the submitted command; it can leave a dangling flag such as +`--updater-options` with no value. Prefer sending an explicit value for each placeholder you use +(or a documented default) instead of dropping the key. See +[Built-in Job Templates](./optimizer-cli-reference.md#built-in-job-templates). + +Built-in Iceberg templates also need an Iceberg Spark runtime on the Spark classpath. They do not +ship that JAR or fill template `jars`, so include it yourself — for example +`"spark.jars":"/path/to/iceberg-spark-runtime-....jar"` inside `spark_conf`. Match the artifact to +your Spark, Scala, and Iceberg versions. Details are under +[Built-in Job Templates](./optimizer-cli-reference.md#built-in-job-templates). + `warehouse_location` may be empty for local filesystem testing. Set it to the warehouse URI for HDFS or cloud object storage. ## Running Against a Local Filesystem diff --git a/docs/table-maintenance-service/optimizer-troubleshooting.md b/docs/table-maintenance-service/optimizer-troubleshooting.md index 3a2111d69d..e9d382e7e0 100644 --- a/docs/table-maintenance-service/optimizer-troubleshooting.md +++ b/docs/table-maintenance-service/optimizer-troubleshooting.md @@ -72,6 +72,22 @@ spark.hadoop.fs.defaultFS=file:/// } ``` +**Built-in Iceberg jobs fail with `Missing Iceberg Spark session extensions`** — +Spark only warns when `IcebergSparkSessionExtensions` is missing, so built-in jobs check the +classpath after `SparkSession` starts and exit with a non-zero status when the Iceberg Spark +runtime is absent. The templates configure Iceberg classes but leave `jars` empty, and +`gravitino-jobs` does not bundle `iceberg-spark-runtime`. A stock Spark install is not enough. +Put a version-matched Iceberg Spark runtime on the job classpath, for example: + +```json +{ + "spark.jars": "/path/to/iceberg-spark-runtime-3.5_2.12-1.11.0.jar" +} +``` + +Use the Spark, Scala, and Iceberg versions that match your cluster. See +[Built-in Job Templates](./optimizer-cli-reference.md#built-in-job-templates). + **Rewrite fails on a multi-level partition** — in release `1.2.0`, rewriting a table partitioned by an identity transform combined with a time transform, such as `PARTITIONED BY (p, days(ts))`, fails with: ```text diff --git a/docs/table-maintenance-service/optimizer.md b/docs/table-maintenance-service/optimizer.md index 22c3362528..4a0acb8d77 100644 --- a/docs/table-maintenance-service/optimizer.md +++ b/docs/table-maintenance-service/optimizer.md @@ -67,6 +67,17 @@ Each step ends with a check. If a check fails, stop there, since every step depe - A running Gravitino server with a metalake. The examples use `test`. - Spark available to the job executor, through either `SPARK_HOME` or `gravitino.jobExecutor.local.sparkHome`. +- An Iceberg Spark runtime on that Spark classpath. Built-in Iceberg templates configure + `IcebergSparkSessionExtensions` and `SparkCatalog`, but `gravitino-jobs` does not ship the + Iceberg Spark runtime and the templates leave `jars` empty so your Spark and Iceberg versions + stay under your control. A stock Spark distribution is not enough. Put a matching + `iceberg-spark-runtime-*` JAR on the job classpath — for example with `spark.jars` in + `spark_conf`, or by installing it into your Spark environment. Pick the artifact that matches + your Spark, Scala, and Iceberg versions. The jobs module is built and tested against Spark 3.5.x + and Iceberg 1.11.0 (for example + `org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.11.0`). Without it, the job fails after + Spark starts with an error naming the missing Iceberg classes. See + [Troubleshooting](./optimizer-troubleshooting.md#job-execution-failures). - `gravitino.job.statusPullIntervalInMs` lowered to `10000` and the server restarted. The default is five minutes, which makes every status check in this walkthrough feel broken. If your Iceberg REST backend runs in memory, do not restart it partway through. Restarting resets both metadata and data files, and you start over. diff --git a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergExpireSnapshotsJob.java b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergExpireSnapshotsJob.java new file mode 100644 index 0000000000..19b9ccf6dd --- /dev/null +++ b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergExpireSnapshotsJob.java @@ -0,0 +1,312 @@ +/* + * 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.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 expiring old snapshots from Iceberg tables. + * + * <p>This job leverages Iceberg's ExpireSnapshots procedure to remove snapshot metadata and + * associated data files that are no longer needed, preventing unbounded metadata growth. + */ +public class IcebergExpireSnapshotsJob implements BuiltInJob { + + private static final String NAME = + JobTemplateProvider.BUILTIN_NAME_PREFIX + "iceberg-expire-snapshots"; + private static final String VERSION = "v1"; + + @Override + public SparkJobTemplate jobTemplate() { + return SparkJobTemplate.builder() + .withName(NAME) + .withComment("Built-in Iceberg expire snapshots job template for metadata cleanup") + .withExecutable(resolveExecutable(IcebergExpireSnapshotsJob.class)) + .withClassName(IcebergExpireSnapshotsJob.class.getName()) + .withArguments(buildArguments()) + .withConfigs(buildSparkConfigs()) + .withCustomFields( + Collections.singletonMap(JobTemplateProvider.PROPERTY_VERSION_KEY, VERSION)) + .build(); + } + + /** + * Main entry point for the expire snapshots 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>--older-than <timestamp> Optional. Expire snapshots older than this timestamp + * (e.g., '2024-01-01 00:00:00') + * <li>--retain-last <count> Optional. Number of most recent snapshots to retain + * <li>--stream-results Optional. Flag to enable streaming of intermediate results + * <li>--spark-conf <spark_conf_json> Optional. JSON map of custom Spark configurations + * </ul> + * + * <p><b>Important Notes on Special Characters:</b> + * + * <ul> + * <li><b>Via Gravitino API:</b> Pass values as-is without shell escaping. Gravitino handles + * escaping internally via ProcessBuilder. + * <li><b>Via Command Line:</b> Use shell quoting. Example: {@code --older-than '2024-01-01 + * 00:00:00'} + * </ul> + * + * <p>Example via command line: --catalog iceberg_catalog --table db.sample --older-than + * '2024-01-01 00:00:00' --retain-last 5 + * + * <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("older_than", "2024-01-01 00:00:00"); + * jobConf.put("retain_last", "5"); + * metalake.runJob("builtin-iceberg-expire-snapshots", jobConf); + * }</pre> + */ + public static void main(String[] args) { + if (args.length < 4) { + printUsage(); + System.exit(1); + } + + // Parse named arguments + Map<String, String> argMap = IcebergJobUtils.parseArguments(args); + + // Validate required arguments + 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); + } + + // Optional arguments + String olderThan = argMap.get("older-than"); + String retainLast = argMap.get("retain-last"); + // --stream-results is a boolean flag (presence = true) + boolean streamResults = argMap.containsKey("stream-results"); + String sparkConfJson = argMap.get("spark-conf"); + + // Validate retain-last if provided + try { + validateRetainLast(retainLast); + } 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 Expire Snapshots"); + + // Apply custom Spark configurations if provided + if (sparkConfJson != null && !sparkConfJson.isEmpty()) { + try { + Map<String, String> customConfigs = IcebergJobUtils.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(); + IcebergJobUtils.requireIcebergSparkRuntimeOrExit(spark); + + try { + // Build the procedure call SQL + String sql = + buildProcedureCall(catalogName, tableIdentifier, olderThan, retainLast, streamResults); + + System.out.println("Executing Iceberg expire_snapshots procedure: " + sql); + + // Execute the procedure + Row[] results = (Row[]) spark.sql(sql).collect(); + + // Print results + if (results.length > 0) { + Row result = results[0]; + System.out.printf( + "Expire Snapshots Results:%n" + + " Deleted data files: %d%n" + + " Deleted manifest files: %d%n" + + " Deleted manifest lists: %d%n", + result.getLong(0), result.getLong(1), result.getLong(2)); + } + + System.out.println("Expire snapshots job completed successfully"); + } catch (Exception e) { + System.err.println("Error executing expire snapshots job: " + e.getMessage()); + e.printStackTrace(); + System.exit(1); + } finally { + spark.stop(); + } + } + + /** + * Build the SQL CALL statement for the expire_snapshots procedure. + * + * @param catalogName Iceberg catalog name + * @param tableIdentifier Fully qualified table name + * @param olderThan Timestamp to expire snapshots older than + * @param retainLast Number of most recent snapshots to retain + * @param streamResults Whether to stream intermediate results + * @return SQL CALL statement + */ + static String buildProcedureCall( + String catalogName, + String tableIdentifier, + String olderThan, + String retainLast, + boolean streamResults) { + + StringBuilder sql = new StringBuilder(); + sql.append("CALL ") + .append(IcebergJobUtils.escapeSqlIdentifier(catalogName)) + .append(".system.expire_snapshots("); + sql.append("table => '").append(IcebergJobUtils.escapeSqlString(tableIdentifier)).append("'"); + + if (olderThan != null && !olderThan.isEmpty()) { + sql.append(", older_than => TIMESTAMP '") + .append(IcebergJobUtils.escapeSqlString(olderThan)) + .append("'"); + } + + if (retainLast != null && !retainLast.isEmpty()) { + sql.append(", retain_last => ").append(Integer.parseInt(retainLast)); + } + + if (streamResults) { + sql.append(", stream_results => true"); + } + + sql.append(")"); + return sql.toString(); + } + + /** + * Validate the retain-last parameter value. + * + * @param retainLast the retain-last value to validate + * @throws IllegalArgumentException if the value is invalid + */ + static void validateRetainLast(String retainLast) { + if (retainLast == null || retainLast.isEmpty()) { + return; // retain-last is optional + } + + try { + int value = Integer.parseInt(retainLast); + if (value < 1) { + throw new IllegalArgumentException( + "Invalid retain-last value '" + retainLast + "'. Must be a positive integer (>= 1)"); + } + } catch (NumberFormatException e) { + throw new IllegalArgumentException( + "Invalid retain-last value '" + retainLast + "'. Must be a positive integer"); + } + } + + /** Print usage information. */ + private static void printUsage() { + System.err.println( + "Usage: IcebergExpireSnapshotsJob [OPTIONS]\n" + + "\n" + + "Required Options:\n" + + " --catalog <name> Iceberg catalog name registered in Spark\n" + + " --table <identifier> Fully qualified table name (e.g., db.table_name)\n" + + "\n" + + "Optional Options:\n" + + " --older-than <timestamp> Expire snapshots older than this timestamp\n" + + " Example: '2024-01-01 00:00:00'\n" + + " Default: 5 days ago (Iceberg default)\n" + + " --retain-last <count> Number of most recent snapshots to retain\n" + + " Must be a positive integer (>= 1)\n" + + " Default: 1 (Iceberg default)\n" + + " --stream-results Enable streaming of intermediate delete results\n" + + " --spark-conf <json> 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 expire with defaults (5 days, retain 1)\n" + + " --catalog iceberg_prod --table db.sample\n" + + "\n" + + " # Expire snapshots older than a specific date\n" + + " --catalog iceberg_prod --table db.sample --older-than '2024-01-01 00:00:00'\n" + + "\n" + + " # Retain the last 5 snapshots\n" + + " --catalog iceberg_prod --table db.sample --retain-last 5\n" + + "\n" + + " # Expire with all options and streaming\n" + + " --catalog iceberg_prod --table db.sample --older-than '2024-06-01 00:00:00' \\\n" + + " --retain-last 3 --stream-results"); + } + + /** + * Build template arguments list with named argument format. + * + * @return list of template arguments + */ + private static List<String> buildArguments() { + return Arrays.asList( + "--catalog", + "{{catalog_name}}", + "--table", + "{{table_identifier}}", + "--older-than", + "{{older_than}}", + "--retain-last", + "{{retain_last}}", + "{{stream_results}}", + "--spark-conf", + "{{spark_conf}}"); + } + + /** + * Build Spark configuration template. + * + * @return map of Spark configuration keys to template values + */ + private static Map<String, String> buildSparkConfigs() { + return IcebergSparkConfigUtils.buildTemplateSparkConfigs(); + } +} 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 new file mode 100644 index 0000000000..9f95e92405 --- /dev/null +++ b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergJobUtils.java @@ -0,0 +1,177 @@ +/* + * 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.HashMap; +import java.util.Map; +import org.apache.gravitino.maintenance.optimizer.common.util.IcebergSparkConfigUtils; +import org.apache.spark.sql.SparkSession; + +/** + * Shared utility methods for Iceberg maintenance jobs. + * + * <p>Provides SQL escaping, argument parsing, Spark configuration utilities, and classpath checks + * used by built-in Iceberg Spark jobs. + */ +public final class IcebergJobUtils { + + private static final String ICEBERG_SPARK_CATALOG = "org.apache.iceberg.spark.SparkCatalog"; + + private IcebergJobUtils() {} + + /** + * Escape single quotes in SQL string literals by replacing ' with ''. + * + * @param value the string value to escape + * @return escaped string safe for use in SQL string literals + */ + public static String escapeSqlString(String value) { + if (value == null) { + return null; + } + return value.replace("'", "''"); + } + + /** + * Escape and quote a SQL identifier with backticks. + * + * <p>Internal backticks are doubled to prevent breaking out of the quoted identifier. The result + * is wrapped in backticks so that identifiers containing special characters (whitespace, dots, + * semicolons) are treated as a single identifier token. + * + * @param identifier the SQL identifier to escape and quote + * @return backtick-quoted identifier safe for use in SQL, or null if input is null + */ + public static String escapeSqlIdentifier(String identifier) { + if (identifier == null) { + return null; + } + String escaped = identifier.replace("`", "``"); + return "`" + escaped + "`"; + } + + /** + * Parse command line arguments in --key value format. + * + * <p>Supports boolean flags (--flag without a value) by storing them with a "true" value. + * + * @param args command line arguments + * @return map of argument names to values + */ + public 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); // Remove "--" prefix + + // Check if there's a value for this key (not another flag) + if (i + 1 < args.length && !args[i + 1].startsWith("--")) { + String value = args[i + 1]; + // Only add non-empty values + if (value != null && !value.trim().isEmpty()) { + argMap.put(key, value); + } + i++; // Skip the value in next iteration + } else { + // Boolean flag with no value - treat as "true" + argMap.put(key, "true"); + } + } + } + + return argMap; + } + + /** + * Parse custom Spark configurations from JSON string. + * + * @param sparkConfJson JSON string containing Spark configurations + * @return map of Spark configuration keys to values + * @throws IllegalArgumentException if JSON parsing fails + */ + public static Map<String, String> parseCustomSparkConfigs(String sparkConfJson) { + return IcebergSparkConfigUtils.parseFlatJsonMap(sparkConfJson, "spark-conf"); + } + + /** + * Ensures the Iceberg Spark runtime is on the current classpath. + * + * <p>Built-in templates configure {@code IcebergSparkSessionExtensions} and {@code SparkCatalog}, + * but Spark only warns when those classes are missing and continues without Iceberg support. Call + * this after {@code SparkSession} creation (so {@code spark.jars} from {@code spark_conf} is + * visible) and fail the job when the runtime is absent. + * + * @throws IllegalStateException when required Iceberg Spark classes cannot be loaded + */ + public static void requireIcebergSparkRuntime() { + requireClass( + IcebergSparkConfigUtils.ICEBERG_SPARK_EXTENSIONS, "Iceberg Spark session extensions"); + requireClass(ICEBERG_SPARK_CATALOG, "Iceberg Spark catalog"); + } + + /** + * Checks the Iceberg Spark runtime after the session is created, then stops Spark and exits the + * process on failure. + * + * @param spark Spark session created for this job; stopped if the runtime check fails + */ + public static void requireIcebergSparkRuntimeOrExit(SparkSession spark) { + try { + requireIcebergSparkRuntime(); + } catch (IllegalStateException e) { + System.err.println("Error: " + e.getMessage()); + spark.stop(); + System.exit(1); + return; + } + } + + /** Visible for unit tests that assert the missing-class error message. */ + static void requireClassForTest(String className, String description) { + requireClass(className, description); + } + + private static void requireClass(String className, String description) { + ClassLoader contextLoader = Thread.currentThread().getContextClassLoader(); + ClassLoader fallbackLoader = IcebergJobUtils.class.getClassLoader(); + try { + // initialize=false: only verify the class is loadable; avoid running <clinit> here and + // misreporting init-time LinkageError as a missing iceberg-spark-runtime jar. + Class.forName(className, false, contextLoader != null ? contextLoader : fallbackLoader); + } catch (ClassNotFoundException | LinkageError first) { + if (contextLoader != null && contextLoader != fallbackLoader) { + try { + Class.forName(className, false, fallbackLoader); + return; + } catch (ClassNotFoundException | LinkageError ignored) { + // Fall through to the user-facing error. + } + } + throw new IllegalStateException( + String.format( + "Missing %s (%s). Built-in Iceberg jobs need iceberg-spark-runtime on the Spark " + + "classpath (for example via spark.jars in spark_conf, or installed into the " + + "Spark environment). A stock Spark distribution does not include it. Match " + + "the artifact to your Spark, Scala, and Iceberg versions.", + description, className), + first); + } + } +} diff --git a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteDataFilesJob.java b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteDataFilesJob.java index c2788281b0..73444783e8 100644 --- a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteDataFilesJob.java +++ b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteDataFilesJob.java @@ -18,11 +18,8 @@ */ package org.apache.gravitino.maintenance.jobs.iceberg; -import com.fasterxml.jackson.core.type.TypeReference; -import com.fasterxml.jackson.databind.ObjectMapper; 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; @@ -160,6 +157,7 @@ public class IcebergRewriteDataFilesJob implements BuiltInJob { } SparkSession spark = sparkBuilder.getOrCreate(); + IcebergJobUtils.requireIcebergSparkRuntimeOrExit(spark); try { // Build the procedure call SQL @@ -426,47 +424,15 @@ public class IcebergRewriteDataFilesJob implements BuiltInJob { } /** - * Parse options from JSON string using Jackson for robust parsing. + * Parse options from a flat JSON map. * - * <p>Expected format: {"key1": "value1", "key2": "value2"} - * - * <p>This method uses Jackson ObjectMapper to properly handle: - * - * <ul> - * <li>Escaped quotes in values - * <li>Colons and commas in values - * <li>Complex JSON structures - * <li>Various data types (strings, numbers, booleans) - * </ul> + * <p>Parse errors name the real CLI flag {@code --options}. Nested objects/arrays are rejected. * * @param optionsJson JSON string * @return map of option keys to values */ static Map<String, String> parseOptionsJson(String optionsJson) { - Map<String, String> options = new HashMap<>(); - if (optionsJson == null || optionsJson.isEmpty()) { - return options; - } - - try { - ObjectMapper mapper = new ObjectMapper(); - // Parse JSON into a Map<String, Object> to handle various value types - Map<String, Object> parsedMap = - mapper.readValue(optionsJson, new TypeReference<Map<String, Object>>() {}); - - // Convert all values to strings - for (Map.Entry<String, Object> entry : parsedMap.entrySet()) { - String key = entry.getKey(); - Object value = entry.getValue(); - // Convert value to string - handles strings, numbers, booleans, etc. - options.put(key, value == null ? "" : value.toString()); - } - } catch (Exception e) { - throw new IllegalArgumentException( - "Failed to parse options JSON: " + optionsJson + ". Error: " + e.getMessage(), e); - } - - return options; + return IcebergSparkConfigUtils.parseFlatJsonMap(optionsJson, "options"); } /** diff --git a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergUpdateStatsAndMetricsJob.java b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergUpdateStatsAndMetricsJob.java index 22a81caf00..7253cb07ec 100644 --- a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergUpdateStatsAndMetricsJob.java +++ b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergUpdateStatsAndMetricsJob.java @@ -96,7 +96,8 @@ public class IcebergUpdateStatsAndMetricsJob implements BuiltInJob { System.exit(1); } - Map<String, String> updaterOptions = parseJsonOptions(argMap.get("updater-options")); + Map<String, String> updaterOptions = + parseJsonOptions(argMap.get("updater-options"), "updater-options"); String sparkConfJson = argMap.get("spark-conf"); SparkSession.Builder sparkBuilder = @@ -110,6 +111,7 @@ public class IcebergUpdateStatsAndMetricsJob implements BuiltInJob { } SparkSession spark = sparkBuilder.getOrCreate(); + IcebergJobUtils.requireIcebergSparkRuntimeOrExit(spark); StatisticsUpdater statisticsUpdater = null; MetricsUpdater metricsUpdater = null; try { @@ -391,15 +393,22 @@ public class IcebergUpdateStatsAndMetricsJob implements BuiltInJob { @VisibleForTesting static Map<String, String> parseCustomSparkConfigs(String sparkConfJson) { - return parseJsonOptions(sparkConfJson); + return parseJsonOptions(sparkConfJson, "spark-conf"); } + /** + * Parse a flat JSON option map and report parse errors with the real CLI flag name. + * + * @param json JSON string; null or empty yields an empty map + * @param optionName CLI flag name without {@code --} (for example {@code updater-options}) + * @return parsed flat string map + */ @VisibleForTesting - static Map<String, String> parseJsonOptions(String json) { + static Map<String, String> parseJsonOptions(String json, String optionName) { if (json == null || json.isEmpty()) { return new HashMap<>(); } - return IcebergSparkConfigUtils.parseFlatJsonMap(json, "json-options"); + return IcebergSparkConfigUtils.parseFlatJsonMap(json, optionName); } @VisibleForTesting diff --git a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergExpireSnapshotsJob.java b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergExpireSnapshotsJob.java new file mode 100644 index 0000000000..90f71478b4 --- /dev/null +++ b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergExpireSnapshotsJob.java @@ -0,0 +1,477 @@ +/* + * 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 static org.junit.jupiter.api.Assertions.fail; + +import java.util.Map; +import org.apache.gravitino.job.JobTemplateProvider; +import org.apache.gravitino.job.SparkJobTemplate; +import org.junit.jupiter.api.Test; + +public class TestIcebergExpireSnapshotsJob { + + @Test + public void testJobTemplateHasCorrectName() { + IcebergExpireSnapshotsJob job = new IcebergExpireSnapshotsJob(); + SparkJobTemplate template = job.jobTemplate(); + + assertNotNull(template); + assertEquals("builtin-iceberg-expire-snapshots", template.name()); + } + + @Test + public void testJobTemplateHasComment() { + IcebergExpireSnapshotsJob job = new IcebergExpireSnapshotsJob(); + SparkJobTemplate template = job.jobTemplate(); + + assertNotNull(template.comment()); + assertFalse(template.comment().trim().isEmpty()); + assertTrue(template.comment().contains("Iceberg")); + } + + @Test + public void testJobTemplateHasExecutable() { + IcebergExpireSnapshotsJob job = new IcebergExpireSnapshotsJob(); + SparkJobTemplate template = job.jobTemplate(); + + assertNotNull(template.executable()); + assertFalse(template.executable().trim().isEmpty()); + } + + @Test + public void testJobTemplateHasMainClass() { + IcebergExpireSnapshotsJob job = new IcebergExpireSnapshotsJob(); + SparkJobTemplate template = job.jobTemplate(); + + assertNotNull(template.className()); + assertEquals(IcebergExpireSnapshotsJob.class.getName(), template.className()); + } + + @Test + public void testJobTemplateHasArguments() { + IcebergExpireSnapshotsJob job = new IcebergExpireSnapshotsJob(); + SparkJobTemplate template = job.jobTemplate(); + + assertNotNull(template.arguments()); + assertEquals(11, template.arguments().size()); + + // 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("--older-than")); + assertTrue(template.arguments().contains("{{older_than}}")); + assertTrue(template.arguments().contains("--retain-last")); + assertTrue(template.arguments().contains("{{retain_last}}")); + // --stream-results is a boolean flag, value is the template variable itself + assertTrue(template.arguments().contains("{{stream_results}}")); + assertTrue(template.arguments().contains("--spark-conf")); + assertTrue(template.arguments().contains("{{spark_conf}}")); + } + + @Test + public void testJobTemplateHasSparkConfigs() { + IcebergExpireSnapshotsJob job = new IcebergExpireSnapshotsJob(); + SparkJobTemplate template = job.jobTemplate(); + + Map<String, String> configs = template.configs(); + assertNotNull(configs); + assertFalse(configs.isEmpty()); + + // Verify Spark runtime configs + assertTrue(configs.containsKey("spark.master")); + assertTrue(configs.containsKey("spark.executor.instances")); + assertTrue(configs.containsKey("spark.executor.cores")); + assertTrue(configs.containsKey("spark.executor.memory")); + assertTrue(configs.containsKey("spark.driver.memory")); + + // Verify Iceberg catalog configs + assertTrue(configs.containsKey("spark.sql.catalog.{{catalog_name}}")); + assertTrue(configs.containsKey("spark.sql.catalog.{{catalog_name}}.type")); + assertTrue(configs.containsKey("spark.sql.catalog.{{catalog_name}}.uri")); + assertTrue(configs.containsKey("spark.sql.catalog.{{catalog_name}}.warehouse")); + + // Verify Iceberg extensions + assertTrue(configs.containsKey("spark.sql.extensions")); + assertEquals( + "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions", + configs.get("spark.sql.extensions")); + } + + @Test + public void testJobTemplateHasVersion() { + IcebergExpireSnapshotsJob job = new IcebergExpireSnapshotsJob(); + SparkJobTemplate template = job.jobTemplate(); + + Map<String, String> customFields = template.customFields(); + assertNotNull(customFields); + assertTrue(customFields.containsKey(JobTemplateProvider.PROPERTY_VERSION_KEY)); + + String version = customFields.get(JobTemplateProvider.PROPERTY_VERSION_KEY); + assertEquals("v1", version); + assertTrue(version.matches(JobTemplateProvider.VERSION_VALUE_PATTERN)); + } + + @Test + public void testJobTemplateNameMatchesBuiltInPattern() { + IcebergExpireSnapshotsJob job = new IcebergExpireSnapshotsJob(); + SparkJobTemplate template = job.jobTemplate(); + + assertTrue(template.name().matches(JobTemplateProvider.BUILTIN_NAME_PATTERN)); + assertTrue(template.name().startsWith(JobTemplateProvider.BUILTIN_NAME_PREFIX)); + } + + // Test parseArguments method + + @Test + public void testParseArgumentsWithAllRequired() { + String[] args = {"--catalog", "iceberg_prod", "--table", "db.sample"}; + Map<String, String> result = IcebergJobUtils.parseArguments(args); + + 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", + "--older-than", "2024-01-01 00:00:00", + "--retain-last", "5" + }; + Map<String, String> result = IcebergJobUtils.parseArguments(args); + + assertEquals(4, result.size()); + assertEquals("iceberg_prod", result.get("catalog")); + assertEquals("db.sample", result.get("table")); + assertEquals("2024-01-01 00:00:00", result.get("older-than")); + assertEquals("5", result.get("retain-last")); + } + + @Test + public void testParseArgumentsWithEmptyValues() { + String[] args = {"--catalog", "iceberg_prod", "--table", "db.sample", "--older-than", ""}; + Map<String, String> result = IcebergJobUtils.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("older-than")); + } + + @Test + public void testParseArgumentsFlagOnly() { + // --stream-results as a flag (no value) should be treated as "true" + String[] args = {"--catalog", "iceberg_prod", "--table", "db.sample", "--stream-results"}; + Map<String, String> result = IcebergJobUtils.parseArguments(args); + + assertEquals(3, result.size()); + assertEquals("iceberg_prod", result.get("catalog")); + assertEquals("db.sample", result.get("table")); + assertEquals("true", result.get("stream-results")); + } + + @Test + public void testParseArgumentsWithAllOptions() { + String[] args = { + "--catalog", + "iceberg_prod", + "--table", + "db.sample", + "--older-than", + "2024-06-01 00:00:00", + "--retain-last", + "3", + "--stream-results", + "--spark-conf", + "{\"spark.executor.memory\":\"4g\"}" + }; + Map<String, String> result = IcebergJobUtils.parseArguments(args); + + assertEquals(6, result.size()); + assertEquals("iceberg_prod", result.get("catalog")); + assertEquals("db.sample", result.get("table")); + assertEquals("2024-06-01 00:00:00", result.get("older-than")); + assertEquals("3", result.get("retain-last")); + assertEquals("true", result.get("stream-results")); + assertEquals("{\"spark.executor.memory\":\"4g\"}", result.get("spark-conf")); + } + + @Test + public void testParseArgumentsOrderIndependent() { + String[] args1 = {"--catalog", "cat1", "--table", "tbl1", "--retain-last", "5"}; + String[] args2 = {"--retain-last", "5", "--table", "tbl1", "--catalog", "cat1"}; + + Map<String, String> result1 = IcebergJobUtils.parseArguments(args1); + Map<String, String> result2 = IcebergJobUtils.parseArguments(args2); + + assertEquals(result1, result2); + } + + // Test buildProcedureCall method + + @Test + public void testBuildProcedureCallMinimal() { + String sql = + IcebergExpireSnapshotsJob.buildProcedureCall( + "iceberg_prod", "db.sample", null, null, false); + + assertEquals("CALL `iceberg_prod`.system.expire_snapshots(table => 'db.sample')", sql); + } + + @Test + public void testBuildProcedureCallWithOlderThan() { + String sql = + IcebergExpireSnapshotsJob.buildProcedureCall( + "iceberg_prod", "db.sample", "2024-01-01 00:00:00", null, false); + + assertEquals( + "CALL `iceberg_prod`.system.expire_snapshots(table => 'db.sample', " + + "older_than => TIMESTAMP '2024-01-01 00:00:00')", + sql); + } + + @Test + public void testBuildProcedureCallWithRetainLast() { + String sql = + IcebergExpireSnapshotsJob.buildProcedureCall("iceberg_prod", "db.sample", null, "5", false); + + assertEquals( + "CALL `iceberg_prod`.system.expire_snapshots(table => 'db.sample', retain_last => 5)", sql); + } + + @Test + public void testBuildProcedureCallWithStreamResults() { + String sql = + IcebergExpireSnapshotsJob.buildProcedureCall("iceberg_prod", "db.sample", null, null, true); + + assertEquals( + "CALL `iceberg_prod`.system.expire_snapshots(table => 'db.sample', " + + "stream_results => true)", + sql); + } + + @Test + public void testBuildProcedureCallWithAllParameters() { + String sql = + IcebergExpireSnapshotsJob.buildProcedureCall( + "iceberg_prod", "db.sample", "2024-01-01 00:00:00", "3", true); + + assertTrue(sql.startsWith("CALL `iceberg_prod`.system.expire_snapshots(")); + assertTrue(sql.contains("table => 'db.sample'")); + assertTrue(sql.contains("older_than => TIMESTAMP '2024-01-01 00:00:00'")); + assertTrue(sql.contains("retain_last => 3")); + assertTrue(sql.contains("stream_results => true")); + assertTrue(sql.endsWith(")")); + } + + @Test + public void testBuildProcedureCallWithEmptyOlderThan() { + String sql = + IcebergExpireSnapshotsJob.buildProcedureCall("iceberg_prod", "db.sample", "", null, false); + + assertEquals("CALL `iceberg_prod`.system.expire_snapshots(table => 'db.sample')", sql); + } + + @Test + public void testBuildProcedureCallWithEmptyRetainLast() { + String sql = + IcebergExpireSnapshotsJob.buildProcedureCall("iceberg_prod", "db.sample", null, "", false); + + assertEquals("CALL `iceberg_prod`.system.expire_snapshots(table => 'db.sample')", sql); + } + + // Test SQL escaping + + @Test + public void testEscapeSqlString() { + // Test basic escaping of single quotes + assertEquals("O''Brien", IcebergJobUtils.escapeSqlString("O'Brien")); + assertEquals("test''with''quotes", IcebergJobUtils.escapeSqlString("test'with'quotes")); + + // Test strings without quotes remain unchanged + assertEquals("normal_string", IcebergJobUtils.escapeSqlString("normal_string")); + + // Test null and empty + assertEquals(null, IcebergJobUtils.escapeSqlString(null)); + assertEquals("", IcebergJobUtils.escapeSqlString("")); + } + + @Test + public void testEscapeSqlIdentifier() { + // Test basic escaping and quoting of backticks + assertEquals("`catalog``name`", IcebergJobUtils.escapeSqlIdentifier("catalog`name")); + + // Test strings without backticks are still quoted + assertEquals("`normal_catalog`", IcebergJobUtils.escapeSqlIdentifier("normal_catalog")); + + // Test null + assertEquals(null, IcebergJobUtils.escapeSqlIdentifier(null)); + } + + @Test + public void testBuildProcedureCallWithSqlInjectionAttempt() { + // Test SQL injection attempt in table name + String maliciousTable = "db.table' OR '1'='1"; + String sql = + IcebergExpireSnapshotsJob.buildProcedureCall( + "iceberg_catalog", maliciousTable, null, null, false); + + // 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 older-than + String maliciousOlderThan = "2024-01-01' OR '1'='1"; + sql = + IcebergExpireSnapshotsJob.buildProcedureCall( + "iceberg_catalog", "db.table", maliciousOlderThan, null, false); + + assertTrue(sql.contains("2024-01-01'' OR ''1''=''1")); + + // Test SQL injection attempt in catalog name + String maliciousCatalog = "catalog`; DROP TABLE users; --"; + sql = + IcebergExpireSnapshotsJob.buildProcedureCall( + maliciousCatalog, "db.table", null, null, false); + + // Verify catalog identifier is quoted and backticks are escaped + assertTrue(sql.contains("`catalog``; DROP TABLE users; --`.system.expire_snapshots")); + } + + @Test + public void testBuildProcedureCallEscapesTableIdentifier() { + String sql = + IcebergExpireSnapshotsJob.buildProcedureCall( + "cat'alog", "db'.table", "2024-01-01' DROP TABLE", null, false); + + // Catalog name should be quoted as an identifier + assertTrue(sql.contains("`cat'alog`")); + // All single quotes in string literals should be escaped + assertTrue(sql.contains("db''.table")); + assertTrue(sql.contains("2024-01-01'' DROP TABLE")); + } + + // Tests for validateRetainLast + + @Test + public void testValidateRetainLastWithValidValue() { + // Should not throw exception + IcebergExpireSnapshotsJob.validateRetainLast("1"); + IcebergExpireSnapshotsJob.validateRetainLast("5"); + IcebergExpireSnapshotsJob.validateRetainLast("100"); + } + + @Test + public void testValidateRetainLastWithNull() { + // Should not throw exception - retain-last is optional + IcebergExpireSnapshotsJob.validateRetainLast(null); + } + + @Test + public void testValidateRetainLastWithEmptyString() { + // Should not throw exception - retain-last is optional + IcebergExpireSnapshotsJob.validateRetainLast(""); + } + + @Test + public void testValidateRetainLastWithZero() { + try { + IcebergExpireSnapshotsJob.validateRetainLast("0"); + fail("Expected IllegalArgumentException for zero retain-last"); + } catch (IllegalArgumentException e) { + assertTrue(e.getMessage().contains("Invalid retain-last value '0'")); + assertTrue(e.getMessage().contains("positive integer")); + } + } + + @Test + public void testValidateRetainLastWithNegative() { + try { + IcebergExpireSnapshotsJob.validateRetainLast("-1"); + fail("Expected IllegalArgumentException for negative retain-last"); + } catch (IllegalArgumentException e) { + assertTrue(e.getMessage().contains("Invalid retain-last value '-1'")); + } + } + + @Test + public void testValidateRetainLastWithNonNumeric() { + try { + IcebergExpireSnapshotsJob.validateRetainLast("abc"); + fail("Expected IllegalArgumentException for non-numeric retain-last"); + } catch (IllegalArgumentException e) { + assertTrue(e.getMessage().contains("Invalid retain-last value 'abc'")); + assertTrue(e.getMessage().contains("positive integer")); + } + } + + // Tests for custom Spark configurations + + @Test + public void testParseCustomSparkConfigsWithValidJson() { + String json = "{\"spark.sql.shuffle.partitions\":\"200\",\"spark.executor.memory\":\"4g\"}"; + Map<String, String> configs = IcebergJobUtils.parseCustomSparkConfigs(json); + + assertEquals(2, configs.size()); + assertEquals("200", configs.get("spark.sql.shuffle.partitions")); + assertEquals("4g", configs.get("spark.executor.memory")); + } + + @Test + public void testParseCustomSparkConfigsWithNumericValues() { + String json = "{\"spark.sql.shuffle.partitions\":200,\"spark.executor.cores\":4}"; + Map<String, String> configs = IcebergJobUtils.parseCustomSparkConfigs(json); + + assertEquals(2, configs.size()); + assertEquals("200", configs.get("spark.sql.shuffle.partitions")); + assertEquals("4", configs.get("spark.executor.cores")); + } + + @Test + public void testParseCustomSparkConfigsWithEmptyString() { + Map<String, String> configs = IcebergJobUtils.parseCustomSparkConfigs(""); + assertTrue(configs.isEmpty()); + } + + @Test + public void testParseCustomSparkConfigsWithNull() { + Map<String, String> configs = IcebergJobUtils.parseCustomSparkConfigs(null); + assertTrue(configs.isEmpty()); + } + + @Test + public void testParseCustomSparkConfigsWithInvalidJson() { + try { + IcebergJobUtils.parseCustomSparkConfigs("{invalid json}"); + fail("Expected IllegalArgumentException for invalid JSON"); + } catch (IllegalArgumentException e) { + assertTrue(e.getMessage().contains("--spark-conf")); + } + } +} diff --git a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergJobUtils.java b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergJobUtils.java new file mode 100644 index 0000000000..5f03f58e3e --- /dev/null +++ b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergJobUtils.java @@ -0,0 +1,57 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.gravitino.maintenance.jobs.iceberg; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import org.junit.jupiter.api.Test; + +public class TestIcebergJobUtils { + + @Test + public void testRequireIcebergSparkRuntimeSucceedsWhenPresent() { + // Test classpath includes iceberg-spark-runtime. + assertDoesNotThrow(IcebergJobUtils::requireIcebergSparkRuntime); + } + + @Test + public void testRequireClassFailsWithActionableMessage() { + IllegalStateException ex = + assertThrows( + IllegalStateException.class, + () -> + IcebergJobUtils.requireClassForTest( + "org.apache.gravitino.does.not.ExistIcebergExtension", + "Iceberg Spark session extensions")); + assertTrue(ex.getMessage().contains("Missing Iceberg Spark session extensions")); + assertTrue(ex.getMessage().contains("iceberg-spark-runtime")); + assertTrue(ex.getMessage().contains("spark.jars")); + } + + @Test + public void testParseCustomSparkConfigsUsesSparkConfFlagName() { + IllegalArgumentException ex = + assertThrows( + IllegalArgumentException.class, + () -> IcebergJobUtils.parseCustomSparkConfigs("{not_json}")); + assertTrue(ex.getMessage().contains("--spark-conf")); + } +} diff --git a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteDataFilesJob.java b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteDataFilesJob.java index 51bb2ba108..c60a689a67 100644 --- a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteDataFilesJob.java +++ b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteDataFilesJob.java @@ -375,7 +375,7 @@ public class TestIcebergRewriteDataFilesJob { IcebergRewriteDataFilesJob.parseOptionsJson(json); fail("Expected IllegalArgumentException for invalid JSON"); } catch (IllegalArgumentException e) { - assertTrue(e.getMessage().contains("Failed to parse options JSON")); + assertTrue(e.getMessage().contains("--options")); } } @@ -705,7 +705,7 @@ public class TestIcebergRewriteDataFilesJob { IcebergRewriteDataFilesJob.parseCustomSparkConfigs(json); fail("Expected IllegalArgumentException for invalid JSON"); } catch (IllegalArgumentException e) { - assertTrue(e.getMessage().contains("Failed to parse Spark configurations JSON")); + assertTrue(e.getMessage().contains("--spark-conf")); } } } diff --git a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergUpdateStatsJob.java b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergUpdateStatsJob.java index 3f0080265e..9fd9307e55 100644 --- a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergUpdateStatsJob.java +++ b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergUpdateStatsJob.java @@ -148,20 +148,43 @@ public class TestIcebergUpdateStatsJob { public void testParseJsonOptions() { Map<String, String> parsed = IcebergUpdateStatsAndMetricsJob.parseJsonOptions( - "{\"a\":\"b\",\"x\":1,\"flag\":true,\"nil\":null}"); + "{\"a\":\"b\",\"x\":1,\"flag\":true,\"nil\":null}", "updater-options"); assertEquals("b", parsed.get("a")); assertEquals("1", parsed.get("x")); assertEquals("true", parsed.get("flag")); assertEquals("", parsed.get("nil")); - assertThrows( - IllegalArgumentException.class, - () -> IcebergUpdateStatsAndMetricsJob.parseJsonOptions("{not_json}")); - assertThrows( - IllegalArgumentException.class, - () -> IcebergUpdateStatsAndMetricsJob.parseJsonOptions("{\"nested\":{\"a\":1}}")); - assertThrows( - IllegalArgumentException.class, - () -> IcebergUpdateStatsAndMetricsJob.parseJsonOptions("{\"array\":[1,2,3]}")); + + IllegalArgumentException invalidJson = + assertThrows( + IllegalArgumentException.class, + () -> + IcebergUpdateStatsAndMetricsJob.parseJsonOptions("{not_json}", "updater-options")); + assertTrue(invalidJson.getMessage().contains("--updater-options")); + + IllegalArgumentException nested = + assertThrows( + IllegalArgumentException.class, + () -> + IcebergUpdateStatsAndMetricsJob.parseJsonOptions( + "{\"nested\":{\"a\":1}}", "updater-options")); + assertTrue(nested.getMessage().contains("--updater-options")); + + IllegalArgumentException array = + assertThrows( + IllegalArgumentException.class, + () -> + IcebergUpdateStatsAndMetricsJob.parseJsonOptions( + "{\"array\":[1,2,3]}", "spark-conf")); + assertTrue(array.getMessage().contains("--spark-conf")); + } + + @Test + public void testParseCustomSparkConfigsUsesSparkConfFlagName() { + IllegalArgumentException ex = + assertThrows( + IllegalArgumentException.class, + () -> IcebergUpdateStatsAndMetricsJob.parseCustomSparkConfigs("{not_json}")); + assertTrue(ex.getMessage().contains("--spark-conf")); } @Test
