This is an automated email from the ASF dual-hosted git repository.
roryqi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new c3c2eadeae [#11194] feat(maintenance): Add
builtin-iceberg-expire-snapshots job (#11206)
c3c2eadeae is described below
commit c3c2eadeaeebf56bfe695ad29b43c8ac9abdd5a0
Author: Akshay Thorat <[email protected]>
AuthorDate: Tue Aug 4 19:11:09 2026 -0700
[#11194] feat(maintenance): Add builtin-iceberg-expire-snapshots job
(#11206)
### What changes were proposed in this pull request?
Add a new built-in Iceberg maintenance job
`builtin-iceberg-expire-snapshots` that expires old snapshots from
Iceberg tables via Spark's `expire_snapshots` procedure.
**Changes:**
- New `IcebergExpireSnapshotsJob` class in `maintenance/jobs` following
the same pattern as `IcebergRewriteDataFilesJob`
- Supports configurable parameters: `older_than` (timestamp),
`retain_last` (number of snapshots to keep), `stream_results` (boolean)
- SQL injection protection via `escapeSqlString()` and
`escapeSqlIdentifier()`
- Input validation for `retain_last` (must be positive integer) and
`stream_results` (must be true/false)
- Registered in `BuiltInJobTemplateProvider`
### Why are the changes needed?
Without periodic snapshot expiration, Iceberg table metadata grows
indefinitely, accumulating snapshot JSON files and manifest lists that
slow down table operations and waste storage. The existing built-in jobs
(`builtin-iceberg-rewrite-data-files` and
`builtin-iceberg-update-stats`) cover data compaction and metrics but do
not address metadata cleanup.
This is one of the most critical Iceberg housekeeping operations. PR
#10500 added Trino-side delegation for `expire_snapshots` as a
procedure, but there is no server-side built-in job that can be
triggered automatically via the Optimizer.
Fix: #11194
**Note:** This PR covers the job layer. The end-to-end policy/strategy
integration (e.g. `IcebergSnapshotExpirationContent`,
`SnapshotExpirationStrategyHandler`) can be added as a follow-up, as
discussed in the issue.
### Does this PR introduce _any_ user-facing change?
No user-facing API changes. Adds a new built-in job template
`builtin-iceberg-expire-snapshots` that will be available for
maintenance job scheduling.
### How was this patch tested?
- Added `TestIcebergExpireSnapshotsJob` with 40 unit tests covering:
- Job template metadata (name, comment, executable, mainClass,
arguments, sparkConfigs, version)
- Argument parsing (required, optional, empty values, missing values,
all options, order independence)
- Procedure call building (minimal, with older-than, retain-last,
stream-results, all params, empty params)
- SQL escaping and injection prevention
- Input validation for retain-last and stream-results
- Custom Spark config parsing (valid JSON, numeric values, empty, null,
invalid JSON)
- All 142 tests in `maintenance:jobs` module pass with 0 failures
---
docs/table-maintenance-service/expire-snapshots.md | 149 +++++++
.../jobs/BuiltInJobTemplateProvider.java | 4 +-
.../jobs/iceberg/IcebergExpireSnapshotsJob.java | 311 ++++++++++++++
.../maintenance/jobs/iceberg/IcebergJobUtils.java | 133 ++++++
.../jobs/iceberg/IcebergRewriteDataFilesJob.java | 107 +----
.../iceberg/TestIcebergExpireSnapshotsJob.java | 477 +++++++++++++++++++++
.../iceberg/TestIcebergRewriteDataFilesJob.java | 41 +-
.../TestIcebergRewriteDataFilesJobWithSpark.java | 2 +-
8 files changed, 1114 insertions(+), 110 deletions(-)
diff --git a/docs/table-maintenance-service/expire-snapshots.md
b/docs/table-maintenance-service/expire-snapshots.md
new file mode 100644
index 0000000000..f486acfd95
--- /dev/null
+++ b/docs/table-maintenance-service/expire-snapshots.md
@@ -0,0 +1,149 @@
+---
+title: "Built-in Expire Snapshots Job"
+slug: /table-maintenance-service/expire-snapshots
+keyword: table maintenance, optimizer, expire snapshots, iceberg, metadata
cleanup
+license: This software is licensed under the Apache License version 2.
+---
+
+## Overview
+
+The `builtin-iceberg-expire-snapshots` job template removes old Iceberg
snapshots and their
+associated metadata files. Without periodic expiration, snapshot JSON files
and manifest lists
+accumulate indefinitely, slowing table operations and wasting storage.
+
+This job executes Iceberg's `expire_snapshots` stored procedure via Spark SQL.
+
+## Job Template
+
+| Property | Value |
+| --- | --- |
+| Name | `builtin-iceberg-expire-snapshots` |
+| Type | Spark |
+| Version | `v1` |
+| Main class |
`org.apache.gravitino.maintenance.jobs.iceberg.IcebergExpireSnapshotsJob` |
+
+## Parameters
+
+### Required
+
+| Key | Description | Example |
+| --- | --- | --- |
+| `catalog_name` | Iceberg catalog name registered in Spark | `rest_catalog` |
+| `table_identifier` | Fully qualified table name | `db.sample` |
+
+### Optional
+
+| Key | Description | Default |
+| --- | --- | --- |
+| `older_than` | Expire snapshots older than this timestamp (`yyyy-MM-dd
HH:mm:ss`) | 5 days ago (Iceberg default) |
+| `retain_last` | Minimum number of most recent snapshots to keep | 1 |
+| `stream_results` | Flag: presence enables streaming of intermediate delete
results | disabled |
+| `spark_conf` | JSON map of custom Spark configurations | none |
+
+## Usage
+
+### Direct job submission via REST
+
+```bash
+curl -X POST -H "Accept: application/vnd.gravitino.v1+json" \
+ -H "Content-Type: application/json" \
+ -d '{
+ "jobTemplateName": "builtin-iceberg-expire-snapshots",
+ "jobConf": {
+ "catalog_name": "rest_catalog",
+ "table_identifier": "db.t1",
+ "older_than": "2024-01-01 00:00:00",
+ "retain_last": "3",
+ "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
+```
+
+### Expire with only `retain_last`
+
+```bash
+curl -X POST -H "Accept: application/vnd.gravitino.v1+json" \
+ -H "Content-Type: application/json" \
+ -d '{
+ "jobTemplateName": "builtin-iceberg-expire-snapshots",
+ "jobConf": {
+ "catalog_name": "rest_catalog",
+ "table_identifier": "db.t1",
+ "retain_last": "5",
+ "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
+```
+
+### Check job status
+
+```bash
+curl -sS "http://localhost:8090/api/metalakes/test/jobs/<job-id>" | jq
+```
+
+## Generated SQL
+
+The job builds and executes a Spark SQL statement:
+
+```sql
+CALL `rest_catalog`.system.expire_snapshots(
+ table => 'db.t1',
+ older_than => TIMESTAMP '2024-01-01 00:00:00',
+ retain_last => 3,
+ stream_results => true
+)
+```
+
+Only non-empty optional parameters are included. The catalog identifier is
backtick-quoted
+for safety.
+
+## Output
+
+On success, the job logs:
+
+```
+Expire Snapshots Results:
+ Deleted data files: 12
+ Deleted manifest files: 8
+ Deleted manifest lists: 3
+```
+
+## Verification
+
+After running the job:
+
+```bash
+# Verify job completed
+curl -sS "http://localhost:8090/api/metalakes/test/jobs/<job-id>" | jq
'.job.state'
+# Expected: "SUCCEEDED"
+
+# Check staging logs
+cat
/tmp/gravitino/jobs/staging/test/builtin-iceberg-expire-snapshots/<job-id>/stdout.log
+```
+
+## Relationship to Other Jobs
+
+| Job | Purpose |
+| --- | --- |
+| `builtin-iceberg-rewrite-data-files` | Compacts small data files |
+| `builtin-iceberg-update-stats` | Collects file statistics and metrics |
+| `builtin-iceberg-expire-snapshots` | Removes old snapshot metadata |
+
+These jobs are complementary. A typical maintenance workflow runs update-stats
first, then
+compaction, then expire-snapshots to clean up the snapshot history created by
compaction.
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 39cc68fd32..3053c02d96 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
@@ -25,6 +25,7 @@ import java.util.regex.Pattern;
import java.util.stream.Collectors;
import org.apache.gravitino.job.JobTemplate;
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.IcebergUpdateStatsAndMetricsJob;
import org.apache.gravitino.maintenance.jobs.spark.SparkPiJob;
@@ -45,7 +46,8 @@ public class BuiltInJobTemplateProvider implements
JobTemplateProvider {
ImmutableList.of(
new SparkPiJob(),
new IcebergRewriteDataFilesJob(),
- new IcebergUpdateStatsAndMetricsJob());
+ new IcebergUpdateStatsAndMetricsJob(),
+ new IcebergExpireSnapshotsJob());
@Override
public List<? extends JobTemplate> jobTemplates() {
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..a7495dae64
--- /dev/null
+++
b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergExpireSnapshotsJob.java
@@ -0,0 +1,311 @@
+/*
+ * 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();
+
+ 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..19cdc2f051
--- /dev/null
+++
b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergJobUtils.java
@@ -0,0 +1,133 @@
+/*
+ * 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 com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Shared utility methods for Iceberg maintenance jobs.
+ *
+ * <p>Provides SQL escaping, argument parsing, and Spark configuration
utilities used by both {@link
+ * IcebergRewriteDataFilesJob} and {@link IcebergExpireSnapshotsJob}.
+ */
+public final class IcebergJobUtils {
+
+ 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) {
+ if (sparkConfJson == null || sparkConfJson.isEmpty()) {
+ return new HashMap<>();
+ }
+
+ try {
+ ObjectMapper mapper = new ObjectMapper();
+ Map<String, Object> parsedMap =
+ mapper.readValue(sparkConfJson, new TypeReference<Map<String,
Object>>() {});
+
+ Map<String, String> configs = new HashMap<>();
+ for (Map.Entry<String, Object> entry : parsedMap.entrySet()) {
+ String key = entry.getKey();
+ Object value = entry.getValue();
+ configs.put(key, value == null ? "" : value.toString());
+ }
+ return configs;
+ } catch (Exception e) {
+ throw new IllegalArgumentException(
+ "Failed to parse Spark configurations JSON: "
+ + sparkConfJson
+ + ". Error: "
+ + e.getMessage(),
+ e);
+ }
+ }
+}
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..6a34c48161 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
@@ -112,7 +112,7 @@ public class IcebergRewriteDataFilesJob implements
BuiltInJob {
}
// Parse named arguments
- Map<String, String> argMap = parseArguments(args);
+ Map<String, String> argMap = IcebergJobUtils.parseArguments(args);
// Validate required arguments
String catalogName = argMap.get("catalog");
@@ -147,7 +147,7 @@ public class IcebergRewriteDataFilesJob implements
BuiltInJob {
// Apply custom Spark configurations if provided
if (sparkConfJson != null && !sparkConfJson.isEmpty()) {
try {
- Map<String, String> customConfigs =
parseCustomSparkConfigs(sparkConfJson);
+ Map<String, String> customConfigs =
IcebergJobUtils.parseCustomSparkConfigs(sparkConfJson);
for (Map.Entry<String, String> entry : customConfigs.entrySet()) {
sparkBuilder.config(entry.getKey(), entry.getValue());
}
@@ -231,20 +231,22 @@ public class IcebergRewriteDataFilesJob implements
BuiltInJob {
StringBuilder sql = new StringBuilder();
sql.append("CALL ")
- .append(escapeSqlIdentifier(catalogName))
+ .append(IcebergJobUtils.escapeSqlIdentifier(catalogName))
.append(".system.rewrite_data_files(");
- sql.append("table =>
'").append(escapeSqlString(tableIdentifier)).append("'");
+ sql.append("table =>
'").append(IcebergJobUtils.escapeSqlString(tableIdentifier)).append("'");
if (strategy != null && !strategy.isEmpty()) {
- sql.append(", strategy =>
'").append(escapeSqlString(strategy)).append("'");
+ sql.append(", strategy =>
'").append(IcebergJobUtils.escapeSqlString(strategy)).append("'");
}
if (sortOrder != null && !sortOrder.isEmpty()) {
- sql.append(", sort_order =>
'").append(escapeSqlString(sortOrder)).append("'");
+ sql.append(", sort_order => '")
+ .append(IcebergJobUtils.escapeSqlString(sortOrder))
+ .append("'");
}
if (whereClause != null && !whereClause.isEmpty()) {
- sql.append(", where =>
'").append(escapeSqlString(whereClause)).append("'");
+ sql.append(", where =>
'").append(IcebergJobUtils.escapeSqlString(whereClause)).append("'");
}
if (optionsJson != null && !optionsJson.isEmpty()) {
@@ -258,9 +260,9 @@ public class IcebergRewriteDataFilesJob implements
BuiltInJob {
sql.append(", ");
}
sql.append("'")
- .append(escapeSqlString(entry.getKey()))
+ .append(IcebergJobUtils.escapeSqlString(entry.getKey()))
.append("', '")
- .append(escapeSqlString(entry.getValue()))
+ .append(IcebergJobUtils.escapeSqlString(entry.getValue()))
.append("'");
first = false;
}
@@ -272,61 +274,19 @@ public class IcebergRewriteDataFilesJob implements
BuiltInJob {
return sql.toString();
}
- /**
- * 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
- */
+ /** Delegates to {@link IcebergJobUtils#escapeSqlString(String)}. */
static String escapeSqlString(String value) {
- if (value == null) {
- return null;
- }
- return value.replace("'", "''");
+ return IcebergJobUtils.escapeSqlString(value);
}
- /**
- * Escape SQL identifiers by replacing backticks and validating format.
- *
- * @param identifier the SQL identifier to escape
- * @return escaped identifier safe for use in SQL
- */
+ /** Delegates to {@link IcebergJobUtils#escapeSqlIdentifier(String)}. */
static String escapeSqlIdentifier(String identifier) {
- if (identifier == null) {
- return null;
- }
- // Replace backticks to prevent breaking out of identifier quotes
- return identifier.replace("`", "``");
+ return IcebergJobUtils.escapeSqlIdentifier(identifier);
}
- /**
- * Parse command line arguments in --key value format.
- *
- * @param args command line arguments
- * @return map of argument names to values
- */
+ /** Delegates to {@link IcebergJobUtils#parseArguments(String[])}. */
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
- 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 {
- System.err.println("Warning: Flag " + args[i] + " has no value,
ignoring");
- }
- }
- }
-
- return argMap;
+ return IcebergJobUtils.parseArguments(args);
}
/**
@@ -352,38 +312,9 @@ public class IcebergRewriteDataFilesJob implements
BuiltInJob {
}
}
- /**
- * 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
- */
+ /** Delegates to {@link IcebergJobUtils#parseCustomSparkConfigs(String)}. */
static Map<String, String> parseCustomSparkConfigs(String sparkConfJson) {
- if (sparkConfJson == null || sparkConfJson.isEmpty()) {
- return new HashMap<>();
- }
-
- try {
- ObjectMapper mapper = new ObjectMapper();
- Map<String, Object> parsedMap =
- mapper.readValue(sparkConfJson, new TypeReference<Map<String,
Object>>() {});
-
- Map<String, String> configs = new HashMap<>();
- for (Map.Entry<String, Object> entry : parsedMap.entrySet()) {
- String key = entry.getKey();
- Object value = entry.getValue();
- configs.put(key, value == null ? "" : value.toString());
- }
- return configs;
- } catch (Exception e) {
- throw new IllegalArgumentException(
- "Failed to parse Spark configurations JSON: "
- + sparkConfJson
- + ". Error: "
- + e.getMessage(),
- e);
- }
+ return IcebergJobUtils.parseCustomSparkConfigs(sparkConfJson);
}
/** Print usage information. */
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..8650e6425e
--- /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("Failed to parse Spark configurations
JSON"));
+ }
+ }
+}
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..20e5f98a80 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
@@ -191,10 +191,10 @@ public class TestIcebergRewriteDataFilesJob {
String[] args = {"--catalog", "iceberg_prod", "--table"};
Map<String, String> result =
IcebergRewriteDataFilesJob.parseArguments(args);
- // Only catalog should be parsed, table has no value
- assertEquals(1, result.size());
+ // catalog has a value, table is a flag at end (treated as boolean)
+ assertEquals(2, result.size());
assertEquals("iceberg_prod", result.get("catalog"));
- assertFalse(result.containsKey("table"));
+ assertEquals("true", result.get("table"));
}
@Test
@@ -397,7 +397,7 @@ public class TestIcebergRewriteDataFilesJob {
IcebergRewriteDataFilesJob.buildProcedureCall(
"iceberg_prod", "db.sample", null, null, null, null);
- assertEquals("CALL iceberg_prod.system.rewrite_data_files(table =>
'db.sample')", sql);
+ assertEquals("CALL `iceberg_prod`.system.rewrite_data_files(table =>
'db.sample')", sql);
}
@Test
@@ -407,7 +407,7 @@ public class TestIcebergRewriteDataFilesJob {
"iceberg_prod", "db.sample", "binpack", null, null, null);
assertEquals(
- "CALL iceberg_prod.system.rewrite_data_files(table => 'db.sample',
strategy => 'binpack')",
+ "CALL `iceberg_prod`.system.rewrite_data_files(table => 'db.sample',
strategy => 'binpack')",
sql);
}
@@ -418,7 +418,7 @@ public class TestIcebergRewriteDataFilesJob {
"iceberg_prod", "db.sample", "sort", "id DESC NULLS LAST", null,
null);
assertEquals(
- "CALL iceberg_prod.system.rewrite_data_files(table => 'db.sample',
strategy => 'sort', sort_order => 'id DESC NULLS LAST')",
+ "CALL `iceberg_prod`.system.rewrite_data_files(table => 'db.sample',
strategy => 'sort', sort_order => 'id DESC NULLS LAST')",
sql);
}
@@ -429,7 +429,7 @@ public class TestIcebergRewriteDataFilesJob {
"iceberg_prod", "db.sample", null, null, "year = 2024", null);
assertEquals(
- "CALL iceberg_prod.system.rewrite_data_files(table => 'db.sample',
where => 'year = 2024')",
+ "CALL `iceberg_prod`.system.rewrite_data_files(table => 'db.sample',
where => 'year = 2024')",
sql);
}
@@ -440,7 +440,8 @@ public class TestIcebergRewriteDataFilesJob {
IcebergRewriteDataFilesJob.buildProcedureCall(
"iceberg_prod", "db.sample", null, null, null, optionsJson);
- assertTrue(sql.startsWith("CALL
iceberg_prod.system.rewrite_data_files(table => 'db.sample'"));
+ assertTrue(
+ sql.startsWith("CALL `iceberg_prod`.system.rewrite_data_files(table =>
'db.sample'"));
assertTrue(sql.contains("options => map("));
assertTrue(sql.contains("'min-input-files', '2'"));
assertTrue(sql.contains("'target-file-size-bytes', '536870912'"));
@@ -454,7 +455,7 @@ public class TestIcebergRewriteDataFilesJob {
IcebergRewriteDataFilesJob.buildProcedureCall(
"iceberg_prod", "db.sample", "sort", "id DESC NULLS LAST", "year =
2024", optionsJson);
- assertTrue(sql.startsWith("CALL iceberg_prod.system.rewrite_data_files("));
+ assertTrue(sql.startsWith("CALL
`iceberg_prod`.system.rewrite_data_files("));
assertTrue(sql.contains("table => 'db.sample'"));
assertTrue(sql.contains("strategy => 'sort'"));
assertTrue(sql.contains("sort_order => 'id DESC NULLS LAST'"));
@@ -469,7 +470,7 @@ public class TestIcebergRewriteDataFilesJob {
IcebergRewriteDataFilesJob.buildProcedureCall(
"iceberg_prod", "db.sample", "", null, null, null);
- assertEquals("CALL iceberg_prod.system.rewrite_data_files(table =>
'db.sample')", sql);
+ assertEquals("CALL `iceberg_prod`.system.rewrite_data_files(table =>
'db.sample')", sql);
}
@Test
@@ -478,7 +479,7 @@ public class TestIcebergRewriteDataFilesJob {
IcebergRewriteDataFilesJob.buildProcedureCall(
"iceberg_prod", "db.sample", null, null, null, "{}");
- assertEquals("CALL iceberg_prod.system.rewrite_data_files(table =>
'db.sample')", sql);
+ assertEquals("CALL `iceberg_prod`.system.rewrite_data_files(table =>
'db.sample')", sql);
}
@Test
@@ -488,7 +489,7 @@ public class TestIcebergRewriteDataFilesJob {
"iceberg_prod", "db.sample", "sort", "zorder(c1,c2,c3)", null,
null);
assertEquals(
- "CALL iceberg_prod.system.rewrite_data_files(table => 'db.sample',
strategy => 'sort', sort_order => 'zorder(c1,c2,c3)')",
+ "CALL `iceberg_prod`.system.rewrite_data_files(table => 'db.sample',
strategy => 'sort', sort_order => 'zorder(c1,c2,c3)')",
sql);
}
@@ -535,12 +536,12 @@ public class TestIcebergRewriteDataFilesJob {
@Test
public void testEscapeSqlIdentifier() {
- // Test basic escaping of backticks
- assertEquals("catalog``name",
IcebergRewriteDataFilesJob.escapeSqlIdentifier("catalog`name"));
+ // Test basic escaping and quoting of backticks
+ assertEquals("`catalog``name`",
IcebergRewriteDataFilesJob.escapeSqlIdentifier("catalog`name"));
- // Test strings without backticks remain unchanged
+ // Test strings without backticks are still quoted
assertEquals(
- "normal_catalog",
IcebergRewriteDataFilesJob.escapeSqlIdentifier("normal_catalog"));
+ "`normal_catalog`",
IcebergRewriteDataFilesJob.escapeSqlIdentifier("normal_catalog"));
// Test null
assertEquals(null, IcebergRewriteDataFilesJob.escapeSqlIdentifier(null));
@@ -572,8 +573,8 @@ public class TestIcebergRewriteDataFilesJob {
IcebergRewriteDataFilesJob.buildProcedureCall(
maliciousCatalog, "db.table", null, null, null, null);
- // Verify backticks are escaped
- assertTrue(sql.contains("catalog``; DROP TABLE users; --"));
+ // Verify catalog identifier is quoted and backticks are escaped
+ assertTrue(sql.contains("`catalog``; DROP TABLE users;
--`.system.rewrite_data_files"));
}
@Test
@@ -582,8 +583,8 @@ public class TestIcebergRewriteDataFilesJob {
IcebergRewriteDataFilesJob.buildProcedureCall(
"cat'alog", "db'.table", "sort'", "id' DESC", "year' = 2024",
"{\"key'\":\"val'ue\"}");
- // Catalog name uses backtick escaping (but no backticks here, so
unchanged)
- assertTrue(sql.contains("cat'alog"));
+ // 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("sort''"));
diff --git
a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteDataFilesJobWithSpark.java
b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteDataFilesJobWithSpark.java
index 2ce3270a20..a0181c2bad 100644
---
a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteDataFilesJobWithSpark.java
+++
b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteDataFilesJobWithSpark.java
@@ -91,7 +91,7 @@ public class TestIcebergRewriteDataFilesJobWithSpark {
catalogName, "db.test_table", null, null, null, null);
assertNotNull(sql);
- assertTrue(sql.startsWith("CALL " + catalogName +
".system.rewrite_data_files("));
+ assertTrue(sql.startsWith("CALL `" + catalogName +
"`.system.rewrite_data_files("));
assertTrue(sql.contains("table => 'db.test_table'"));
}
