Copilot commented on code in PR #11206: URL: https://github.com/apache/gravitino/pull/11206#discussion_r3614088322
########## maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergExpireSnapshotsJob.java: ########## @@ -0,0 +1,506 @@ +/* + * 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(12, template.arguments().size()); // 6 flags * 2 (flag + value) + + // Verify all expected arguments are present + assertTrue(template.arguments().contains("--catalog")); + assertTrue(template.arguments().contains("{{catalog_name}}")); + assertTrue(template.arguments().contains("--table")); + assertTrue(template.arguments().contains("{{table_identifier}}")); + assertTrue(template.arguments().contains("--older-than")); + assertTrue(template.arguments().contains("{{older_than}}")); + assertTrue(template.arguments().contains("--retain-last")); + assertTrue(template.arguments().contains("{{retain_last}}")); + assertTrue(template.arguments().contains("--stream-results")); + 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 = IcebergExpireSnapshotsJob.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 = IcebergExpireSnapshotsJob.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 = IcebergExpireSnapshotsJob.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 testParseArgumentsWithMissingValues() { + String[] args = {"--catalog", "iceberg_prod", "--table"}; + Map<String, String> result = IcebergExpireSnapshotsJob.parseArguments(args); + + // Only catalog should be parsed, table has no value + assertEquals(1, result.size()); + assertEquals("iceberg_prod", result.get("catalog")); + assertFalse(result.containsKey("table")); + } + + @Test + public void testParseArgumentsWithAllOptions() { + String[] args = { + "--catalog", "iceberg_prod", + "--table", "db.sample", + "--older-than", "2024-06-01 00:00:00", + "--retain-last", "3", + "--stream-results", "true", + "--spark-conf", "{\"spark.executor.memory\":\"4g\"}" + }; + Map<String, String> result = IcebergExpireSnapshotsJob.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 = IcebergExpireSnapshotsJob.parseArguments(args1); + Map<String, String> result2 = IcebergExpireSnapshotsJob.parseArguments(args2); + + assertEquals(result1, result2); + } + + // Test buildProcedureCall method + + @Test + public void testBuildProcedureCallMinimal() { + String sql = + IcebergExpireSnapshotsJob.buildProcedureCall("iceberg_prod", "db.sample", null, null, null); + + 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, null); + + 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", null); + + 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, null); + + assertEquals("CALL iceberg_prod.system.expire_snapshots(table => 'db.sample')", sql); + } + + @Test + public void testBuildProcedureCallWithEmptyRetainLast() { + String sql = + IcebergExpireSnapshotsJob.buildProcedureCall("iceberg_prod", "db.sample", null, "", null); + + 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", IcebergExpireSnapshotsJob.escapeSqlString("O'Brien")); + assertEquals( + "test''with''quotes", IcebergExpireSnapshotsJob.escapeSqlString("test'with'quotes")); + + // Test strings without quotes remain unchanged + assertEquals("normal_string", IcebergExpireSnapshotsJob.escapeSqlString("normal_string")); + + // Test null and empty + assertEquals(null, IcebergExpireSnapshotsJob.escapeSqlString(null)); + assertEquals("", IcebergExpireSnapshotsJob.escapeSqlString("")); + } + + @Test + public void testEscapeSqlIdentifier() { + // Test basic escaping of backticks + assertEquals("catalog``name", IcebergExpireSnapshotsJob.escapeSqlIdentifier("catalog`name")); + + // Test strings without backticks remain unchanged + assertEquals("normal_catalog", IcebergExpireSnapshotsJob.escapeSqlIdentifier("normal_catalog")); + Review Comment: These assertions encode the current (unquoted) identifier escaping behavior. If `escapeSqlIdentifier()` is updated to return a backtick-quoted identifier for safety, these expectations should be updated accordingly (including the catalog-name injection assertion). ########## maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergExpireSnapshotsJob.java: ########## @@ -0,0 +1,430 @@ +/* + * 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.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import org.apache.gravitino.job.JobTemplateProvider; +import org.apache.gravitino.job.SparkJobTemplate; +import org.apache.gravitino.maintenance.jobs.BuiltInJob; +import org.apache.gravitino.maintenance.optimizer.common.util.IcebergSparkConfigUtils; +import org.apache.spark.sql.Row; +import org.apache.spark.sql.SparkSession; + +/** + * Built-in job for 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 <boolean> Optional. Whether to stream 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 = 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"); + String streamResults = argMap.get("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); + } + + // Validate stream-results if provided + try { + validateStreamResults(streamResults); + } 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 = 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, + String streamResults) { + + StringBuilder sql = new StringBuilder(); + sql.append("CALL ") + .append(escapeSqlIdentifier(catalogName)) + .append(".system.expire_snapshots("); + sql.append("table => '").append(escapeSqlString(tableIdentifier)).append("'"); + + if (olderThan != null && !olderThan.isEmpty()) { + sql.append(", older_than => TIMESTAMP '").append(escapeSqlString(olderThan)).append("'"); + } + + if (retainLast != null && !retainLast.isEmpty()) { + sql.append(", retain_last => ").append(Integer.parseInt(retainLast)); + } + + if (streamResults != null && !streamResults.isEmpty()) { + sql.append(", stream_results => ").append(Boolean.parseBoolean(streamResults)); + } + + sql.append(")"); + 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 + */ + static String escapeSqlString(String value) { + if (value == null) { + return null; + } + return value.replace("'", "''"); + } + + /** + * Escape SQL identifiers by replacing backticks and validating format. + * + * @param identifier the SQL identifier to escape + * @return escaped identifier safe for use in SQL + */ + static String escapeSqlIdentifier(String identifier) { + if (identifier == null) { + return null; + } + // Replace backticks to prevent breaking out of identifier quotes + return identifier.replace("`", "``"); + } Review Comment: `escapeSqlIdentifier()` only doubles backticks but does not actually quote the identifier. Since `buildProcedureCall()` concatenates the catalog name directly into `CALL <catalog>.system...`, a catalog value containing whitespace/dots/semicolons can still break the statement (and undermines the claimed SQL injection protection). Consider quoting the identifier with backticks (matching `IcebergUpdateStatsAndMetricsJob.escapeSqlIdentifier`) so the entire catalog name is treated as a single identifier. ########## maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergExpireSnapshotsJob.java: ########## @@ -0,0 +1,506 @@ +/* + * 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(12, template.arguments().size()); // 6 flags * 2 (flag + value) + + // Verify all expected arguments are present + assertTrue(template.arguments().contains("--catalog")); + assertTrue(template.arguments().contains("{{catalog_name}}")); + assertTrue(template.arguments().contains("--table")); + assertTrue(template.arguments().contains("{{table_identifier}}")); + assertTrue(template.arguments().contains("--older-than")); + assertTrue(template.arguments().contains("{{older_than}}")); + assertTrue(template.arguments().contains("--retain-last")); + assertTrue(template.arguments().contains("{{retain_last}}")); + assertTrue(template.arguments().contains("--stream-results")); + 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 = IcebergExpireSnapshotsJob.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 = IcebergExpireSnapshotsJob.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 = IcebergExpireSnapshotsJob.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 testParseArgumentsWithMissingValues() { + String[] args = {"--catalog", "iceberg_prod", "--table"}; + Map<String, String> result = IcebergExpireSnapshotsJob.parseArguments(args); + + // Only catalog should be parsed, table has no value + assertEquals(1, result.size()); + assertEquals("iceberg_prod", result.get("catalog")); + assertFalse(result.containsKey("table")); + } + + @Test + public void testParseArgumentsWithAllOptions() { + String[] args = { + "--catalog", "iceberg_prod", + "--table", "db.sample", + "--older-than", "2024-06-01 00:00:00", + "--retain-last", "3", + "--stream-results", "true", + "--spark-conf", "{\"spark.executor.memory\":\"4g\"}" + }; + Map<String, String> result = IcebergExpireSnapshotsJob.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 = IcebergExpireSnapshotsJob.parseArguments(args1); + Map<String, String> result2 = IcebergExpireSnapshotsJob.parseArguments(args2); + + assertEquals(result1, result2); + } + + // Test buildProcedureCall method + + @Test + public void testBuildProcedureCallMinimal() { + String sql = + IcebergExpireSnapshotsJob.buildProcedureCall("iceberg_prod", "db.sample", null, null, null); + + 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, null); + + 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", null); + + 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, null); + + assertEquals("CALL iceberg_prod.system.expire_snapshots(table => 'db.sample')", sql); + } + + @Test + public void testBuildProcedureCallWithEmptyRetainLast() { + String sql = + IcebergExpireSnapshotsJob.buildProcedureCall("iceberg_prod", "db.sample", null, "", null); + + 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", IcebergExpireSnapshotsJob.escapeSqlString("O'Brien")); + assertEquals( + "test''with''quotes", IcebergExpireSnapshotsJob.escapeSqlString("test'with'quotes")); + + // Test strings without quotes remain unchanged + assertEquals("normal_string", IcebergExpireSnapshotsJob.escapeSqlString("normal_string")); + + // Test null and empty + assertEquals(null, IcebergExpireSnapshotsJob.escapeSqlString(null)); + assertEquals("", IcebergExpireSnapshotsJob.escapeSqlString("")); + } + + @Test + public void testEscapeSqlIdentifier() { + // Test basic escaping of backticks + assertEquals("catalog``name", IcebergExpireSnapshotsJob.escapeSqlIdentifier("catalog`name")); + + // Test strings without backticks remain unchanged + assertEquals("normal_catalog", IcebergExpireSnapshotsJob.escapeSqlIdentifier("normal_catalog")); + + // Test null + assertEquals(null, IcebergExpireSnapshotsJob.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, null); + + // Verify single quotes are escaped (becomes '') + assertTrue(sql.contains("db.table'' OR ''1''=''1")); + assertFalse(sql.contains("' OR '1'='1")); + + // Test SQL injection attempt in older-than + String maliciousOlderThan = "2024-01-01' OR '1'='1"; + sql = + IcebergExpireSnapshotsJob.buildProcedureCall( + "iceberg_catalog", "db.table", maliciousOlderThan, null, null); + + 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, null); + + // Verify backticks are escaped + assertTrue(sql.contains("catalog``; DROP TABLE users; --")); + } Review Comment: If the catalog identifier is backtick-quoted, the injection test should assert for the quoted form as well; otherwise the test can pass even though the statement is still syntactically injectable via unquoted identifier concatenation. ########## maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergExpireSnapshotsJob.java: ########## @@ -0,0 +1,506 @@ +/* + * 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(12, template.arguments().size()); // 6 flags * 2 (flag + value) + + // Verify all expected arguments are present + assertTrue(template.arguments().contains("--catalog")); + assertTrue(template.arguments().contains("{{catalog_name}}")); + assertTrue(template.arguments().contains("--table")); + assertTrue(template.arguments().contains("{{table_identifier}}")); + assertTrue(template.arguments().contains("--older-than")); + assertTrue(template.arguments().contains("{{older_than}}")); + assertTrue(template.arguments().contains("--retain-last")); + assertTrue(template.arguments().contains("{{retain_last}}")); + assertTrue(template.arguments().contains("--stream-results")); + 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 = IcebergExpireSnapshotsJob.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 = IcebergExpireSnapshotsJob.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 = IcebergExpireSnapshotsJob.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 testParseArgumentsWithMissingValues() { + String[] args = {"--catalog", "iceberg_prod", "--table"}; + Map<String, String> result = IcebergExpireSnapshotsJob.parseArguments(args); + + // Only catalog should be parsed, table has no value + assertEquals(1, result.size()); + assertEquals("iceberg_prod", result.get("catalog")); + assertFalse(result.containsKey("table")); + } + + @Test + public void testParseArgumentsWithAllOptions() { + String[] args = { + "--catalog", "iceberg_prod", + "--table", "db.sample", + "--older-than", "2024-06-01 00:00:00", + "--retain-last", "3", + "--stream-results", "true", + "--spark-conf", "{\"spark.executor.memory\":\"4g\"}" + }; + Map<String, String> result = IcebergExpireSnapshotsJob.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 = IcebergExpireSnapshotsJob.parseArguments(args1); + Map<String, String> result2 = IcebergExpireSnapshotsJob.parseArguments(args2); + + assertEquals(result1, result2); + } + + // Test buildProcedureCall method + + @Test + public void testBuildProcedureCallMinimal() { + String sql = + IcebergExpireSnapshotsJob.buildProcedureCall("iceberg_prod", "db.sample", null, null, null); + + assertEquals("CALL iceberg_prod.system.expire_snapshots(table => 'db.sample')", sql); + } Review Comment: If the catalog identifier is backtick-quoted for safety (recommended), this expected SQL should include the quoted catalog name as well. ########## maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergExpireSnapshotsJob.java: ########## @@ -0,0 +1,506 @@ +/* + * 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(12, template.arguments().size()); // 6 flags * 2 (flag + value) + + // Verify all expected arguments are present + assertTrue(template.arguments().contains("--catalog")); + assertTrue(template.arguments().contains("{{catalog_name}}")); + assertTrue(template.arguments().contains("--table")); + assertTrue(template.arguments().contains("{{table_identifier}}")); + assertTrue(template.arguments().contains("--older-than")); + assertTrue(template.arguments().contains("{{older_than}}")); + assertTrue(template.arguments().contains("--retain-last")); + assertTrue(template.arguments().contains("{{retain_last}}")); + assertTrue(template.arguments().contains("--stream-results")); + 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 = IcebergExpireSnapshotsJob.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 = IcebergExpireSnapshotsJob.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 = IcebergExpireSnapshotsJob.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 testParseArgumentsWithMissingValues() { + String[] args = {"--catalog", "iceberg_prod", "--table"}; + Map<String, String> result = IcebergExpireSnapshotsJob.parseArguments(args); + + // Only catalog should be parsed, table has no value + assertEquals(1, result.size()); + assertEquals("iceberg_prod", result.get("catalog")); + assertFalse(result.containsKey("table")); + } + + @Test + public void testParseArgumentsWithAllOptions() { + String[] args = { + "--catalog", "iceberg_prod", + "--table", "db.sample", + "--older-than", "2024-06-01 00:00:00", + "--retain-last", "3", + "--stream-results", "true", + "--spark-conf", "{\"spark.executor.memory\":\"4g\"}" + }; + Map<String, String> result = IcebergExpireSnapshotsJob.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 = IcebergExpireSnapshotsJob.parseArguments(args1); + Map<String, String> result2 = IcebergExpireSnapshotsJob.parseArguments(args2); + + assertEquals(result1, result2); + } + + // Test buildProcedureCall method + + @Test + public void testBuildProcedureCallMinimal() { + String sql = + IcebergExpireSnapshotsJob.buildProcedureCall("iceberg_prod", "db.sample", null, null, null); + + 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, null); + + 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", null); + + 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); Review Comment: If the catalog identifier becomes backtick-quoted for safety, update this expected SQL to include the quoted catalog name. ########## maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergExpireSnapshotsJob.java: ########## @@ -0,0 +1,506 @@ +/* + * 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(12, template.arguments().size()); // 6 flags * 2 (flag + value) + + // Verify all expected arguments are present + assertTrue(template.arguments().contains("--catalog")); + assertTrue(template.arguments().contains("{{catalog_name}}")); + assertTrue(template.arguments().contains("--table")); + assertTrue(template.arguments().contains("{{table_identifier}}")); + assertTrue(template.arguments().contains("--older-than")); + assertTrue(template.arguments().contains("{{older_than}}")); + assertTrue(template.arguments().contains("--retain-last")); + assertTrue(template.arguments().contains("{{retain_last}}")); + assertTrue(template.arguments().contains("--stream-results")); + 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 = IcebergExpireSnapshotsJob.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 = IcebergExpireSnapshotsJob.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 = IcebergExpireSnapshotsJob.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 testParseArgumentsWithMissingValues() { + String[] args = {"--catalog", "iceberg_prod", "--table"}; + Map<String, String> result = IcebergExpireSnapshotsJob.parseArguments(args); + + // Only catalog should be parsed, table has no value + assertEquals(1, result.size()); + assertEquals("iceberg_prod", result.get("catalog")); + assertFalse(result.containsKey("table")); + } + + @Test + public void testParseArgumentsWithAllOptions() { + String[] args = { + "--catalog", "iceberg_prod", + "--table", "db.sample", + "--older-than", "2024-06-01 00:00:00", + "--retain-last", "3", + "--stream-results", "true", + "--spark-conf", "{\"spark.executor.memory\":\"4g\"}" + }; + Map<String, String> result = IcebergExpireSnapshotsJob.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 = IcebergExpireSnapshotsJob.parseArguments(args1); + Map<String, String> result2 = IcebergExpireSnapshotsJob.parseArguments(args2); + + assertEquals(result1, result2); + } + + // Test buildProcedureCall method + + @Test + public void testBuildProcedureCallMinimal() { + String sql = + IcebergExpireSnapshotsJob.buildProcedureCall("iceberg_prod", "db.sample", null, null, null); + + 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, null); + + assertEquals( + "CALL iceberg_prod.system.expire_snapshots(table => 'db.sample', " + + "older_than => TIMESTAMP '2024-01-01 00:00:00')", + sql); Review Comment: If the catalog identifier becomes backtick-quoted for safety, update this expected SQL to include the quoted catalog name. ########## maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergExpireSnapshotsJob.java: ########## @@ -0,0 +1,506 @@ +/* + * 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(12, template.arguments().size()); // 6 flags * 2 (flag + value) + + // Verify all expected arguments are present + assertTrue(template.arguments().contains("--catalog")); + assertTrue(template.arguments().contains("{{catalog_name}}")); + assertTrue(template.arguments().contains("--table")); + assertTrue(template.arguments().contains("{{table_identifier}}")); + assertTrue(template.arguments().contains("--older-than")); + assertTrue(template.arguments().contains("{{older_than}}")); + assertTrue(template.arguments().contains("--retain-last")); + assertTrue(template.arguments().contains("{{retain_last}}")); + assertTrue(template.arguments().contains("--stream-results")); + 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 = IcebergExpireSnapshotsJob.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 = IcebergExpireSnapshotsJob.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 = IcebergExpireSnapshotsJob.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 testParseArgumentsWithMissingValues() { + String[] args = {"--catalog", "iceberg_prod", "--table"}; + Map<String, String> result = IcebergExpireSnapshotsJob.parseArguments(args); + + // Only catalog should be parsed, table has no value + assertEquals(1, result.size()); + assertEquals("iceberg_prod", result.get("catalog")); + assertFalse(result.containsKey("table")); + } + + @Test + public void testParseArgumentsWithAllOptions() { + String[] args = { + "--catalog", "iceberg_prod", + "--table", "db.sample", + "--older-than", "2024-06-01 00:00:00", + "--retain-last", "3", + "--stream-results", "true", + "--spark-conf", "{\"spark.executor.memory\":\"4g\"}" + }; + Map<String, String> result = IcebergExpireSnapshotsJob.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 = IcebergExpireSnapshotsJob.parseArguments(args1); + Map<String, String> result2 = IcebergExpireSnapshotsJob.parseArguments(args2); + + assertEquals(result1, result2); + } + + // Test buildProcedureCall method + + @Test + public void testBuildProcedureCallMinimal() { + String sql = + IcebergExpireSnapshotsJob.buildProcedureCall("iceberg_prod", "db.sample", null, null, null); + + 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, null); + + 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", null); + + 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, null); + + assertEquals("CALL iceberg_prod.system.expire_snapshots(table => 'db.sample')", sql); Review Comment: If the catalog identifier becomes backtick-quoted for safety, update this expected SQL to include the quoted catalog name. ########## maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergExpireSnapshotsJob.java: ########## @@ -0,0 +1,506 @@ +/* + * 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(12, template.arguments().size()); // 6 flags * 2 (flag + value) + + // Verify all expected arguments are present + assertTrue(template.arguments().contains("--catalog")); + assertTrue(template.arguments().contains("{{catalog_name}}")); + assertTrue(template.arguments().contains("--table")); + assertTrue(template.arguments().contains("{{table_identifier}}")); + assertTrue(template.arguments().contains("--older-than")); + assertTrue(template.arguments().contains("{{older_than}}")); + assertTrue(template.arguments().contains("--retain-last")); + assertTrue(template.arguments().contains("{{retain_last}}")); + assertTrue(template.arguments().contains("--stream-results")); + 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 = IcebergExpireSnapshotsJob.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 = IcebergExpireSnapshotsJob.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 = IcebergExpireSnapshotsJob.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 testParseArgumentsWithMissingValues() { + String[] args = {"--catalog", "iceberg_prod", "--table"}; + Map<String, String> result = IcebergExpireSnapshotsJob.parseArguments(args); + + // Only catalog should be parsed, table has no value + assertEquals(1, result.size()); + assertEquals("iceberg_prod", result.get("catalog")); + assertFalse(result.containsKey("table")); + } + + @Test + public void testParseArgumentsWithAllOptions() { + String[] args = { + "--catalog", "iceberg_prod", + "--table", "db.sample", + "--older-than", "2024-06-01 00:00:00", + "--retain-last", "3", + "--stream-results", "true", + "--spark-conf", "{\"spark.executor.memory\":\"4g\"}" + }; + Map<String, String> result = IcebergExpireSnapshotsJob.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 = IcebergExpireSnapshotsJob.parseArguments(args1); + Map<String, String> result2 = IcebergExpireSnapshotsJob.parseArguments(args2); + + assertEquals(result1, result2); + } + + // Test buildProcedureCall method + + @Test + public void testBuildProcedureCallMinimal() { + String sql = + IcebergExpireSnapshotsJob.buildProcedureCall("iceberg_prod", "db.sample", null, null, null); + + 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, null); + + 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", null); + + 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, null); + + assertEquals("CALL iceberg_prod.system.expire_snapshots(table => 'db.sample')", sql); + } + + @Test + public void testBuildProcedureCallWithEmptyRetainLast() { + String sql = + IcebergExpireSnapshotsJob.buildProcedureCall("iceberg_prod", "db.sample", null, "", null); + + assertEquals("CALL iceberg_prod.system.expire_snapshots(table => 'db.sample')", sql); Review Comment: If the catalog identifier becomes backtick-quoted for safety, update this expected SQL to include the quoted catalog name. ########## maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergExpireSnapshotsJob.java: ########## @@ -0,0 +1,506 @@ +/* + * 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(12, template.arguments().size()); // 6 flags * 2 (flag + value) + + // Verify all expected arguments are present + assertTrue(template.arguments().contains("--catalog")); + assertTrue(template.arguments().contains("{{catalog_name}}")); + assertTrue(template.arguments().contains("--table")); + assertTrue(template.arguments().contains("{{table_identifier}}")); + assertTrue(template.arguments().contains("--older-than")); + assertTrue(template.arguments().contains("{{older_than}}")); + assertTrue(template.arguments().contains("--retain-last")); + assertTrue(template.arguments().contains("{{retain_last}}")); + assertTrue(template.arguments().contains("--stream-results")); + 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 = IcebergExpireSnapshotsJob.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 = IcebergExpireSnapshotsJob.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 = IcebergExpireSnapshotsJob.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 testParseArgumentsWithMissingValues() { + String[] args = {"--catalog", "iceberg_prod", "--table"}; + Map<String, String> result = IcebergExpireSnapshotsJob.parseArguments(args); + + // Only catalog should be parsed, table has no value + assertEquals(1, result.size()); + assertEquals("iceberg_prod", result.get("catalog")); + assertFalse(result.containsKey("table")); + } + + @Test + public void testParseArgumentsWithAllOptions() { + String[] args = { + "--catalog", "iceberg_prod", + "--table", "db.sample", + "--older-than", "2024-06-01 00:00:00", + "--retain-last", "3", + "--stream-results", "true", + "--spark-conf", "{\"spark.executor.memory\":\"4g\"}" + }; + Map<String, String> result = IcebergExpireSnapshotsJob.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 = IcebergExpireSnapshotsJob.parseArguments(args1); + Map<String, String> result2 = IcebergExpireSnapshotsJob.parseArguments(args2); + + assertEquals(result1, result2); + } + + // Test buildProcedureCall method + + @Test + public void testBuildProcedureCallMinimal() { + String sql = + IcebergExpireSnapshotsJob.buildProcedureCall("iceberg_prod", "db.sample", null, null, null); + + 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, null); + + 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", null); + + 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, null); + + assertEquals("CALL iceberg_prod.system.expire_snapshots(table => 'db.sample')", sql); + } + + @Test + public void testBuildProcedureCallWithEmptyRetainLast() { + String sql = + IcebergExpireSnapshotsJob.buildProcedureCall("iceberg_prod", "db.sample", null, "", null); + + 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", IcebergExpireSnapshotsJob.escapeSqlString("O'Brien")); + assertEquals( + "test''with''quotes", IcebergExpireSnapshotsJob.escapeSqlString("test'with'quotes")); + + // Test strings without quotes remain unchanged + assertEquals("normal_string", IcebergExpireSnapshotsJob.escapeSqlString("normal_string")); + + // Test null and empty + assertEquals(null, IcebergExpireSnapshotsJob.escapeSqlString(null)); + assertEquals("", IcebergExpireSnapshotsJob.escapeSqlString("")); + } + + @Test + public void testEscapeSqlIdentifier() { + // Test basic escaping of backticks + assertEquals("catalog``name", IcebergExpireSnapshotsJob.escapeSqlIdentifier("catalog`name")); + + // Test strings without backticks remain unchanged + assertEquals("normal_catalog", IcebergExpireSnapshotsJob.escapeSqlIdentifier("normal_catalog")); + + // Test null + assertEquals(null, IcebergExpireSnapshotsJob.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, null); + + // Verify single quotes are escaped (becomes '') + assertTrue(sql.contains("db.table'' OR ''1''=''1")); + assertFalse(sql.contains("' OR '1'='1")); + + // Test SQL injection attempt in older-than + String maliciousOlderThan = "2024-01-01' OR '1'='1"; + sql = + IcebergExpireSnapshotsJob.buildProcedureCall( + "iceberg_catalog", "db.table", maliciousOlderThan, null, null); + + 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, null); + + // Verify backticks are escaped + assertTrue(sql.contains("catalog``; DROP TABLE users; --")); + } + + @Test + public void testBuildProcedureCallEscapesTableIdentifier() { + String sql = + IcebergExpireSnapshotsJob.buildProcedureCall( + "cat'alog", "db'.table", "2024-01-01' DROP TABLE", null, null); + + // Catalog name uses backtick escaping (but no backticks here, so unchanged) + assertTrue(sql.contains("cat'alog")); Review Comment: If `escapeSqlIdentifier()` is updated to return a backtick-quoted identifier, this assertion should expect the quoted catalog name (including any special characters). ########## maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergExpireSnapshotsJob.java: ########## @@ -0,0 +1,430 @@ +/* + * 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.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import org.apache.gravitino.job.JobTemplateProvider; +import org.apache.gravitino.job.SparkJobTemplate; +import org.apache.gravitino.maintenance.jobs.BuiltInJob; +import org.apache.gravitino.maintenance.optimizer.common.util.IcebergSparkConfigUtils; +import org.apache.spark.sql.Row; +import org.apache.spark.sql.SparkSession; + +/** + * Built-in job for 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 <boolean> Optional. Whether to stream 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 = 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"); + String streamResults = argMap.get("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); + } + + // Validate stream-results if provided + try { + validateStreamResults(streamResults); + } 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 = 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, + String streamResults) { + + StringBuilder sql = new StringBuilder(); + sql.append("CALL ") + .append(escapeSqlIdentifier(catalogName)) + .append(".system.expire_snapshots("); + sql.append("table => '").append(escapeSqlString(tableIdentifier)).append("'"); + + if (olderThan != null && !olderThan.isEmpty()) { + sql.append(", older_than => TIMESTAMP '").append(escapeSqlString(olderThan)).append("'"); + } + + if (retainLast != null && !retainLast.isEmpty()) { + sql.append(", retain_last => ").append(Integer.parseInt(retainLast)); + } + + if (streamResults != null && !streamResults.isEmpty()) { + sql.append(", stream_results => ").append(Boolean.parseBoolean(streamResults)); + } + + sql.append(")"); + 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 + */ + static String escapeSqlString(String value) { + if (value == null) { + return null; + } + return value.replace("'", "''"); + } + + /** + * Escape SQL identifiers by replacing backticks and validating format. + * + * @param identifier the SQL identifier to escape + * @return escaped identifier safe for use in SQL + */ + static String escapeSqlIdentifier(String identifier) { + if (identifier == null) { + return null; + } + // Replace backticks to prevent breaking out of identifier quotes + return identifier.replace("`", "``"); + } + + /** + * Parse command line arguments in --key value format. + * + * @param args command line arguments + * @return map of argument names to values + */ + static Map<String, String> parseArguments(String[] args) { + Map<String, String> argMap = new HashMap<>(); + + for (int i = 0; i < args.length; i++) { + if (args[i].startsWith("--")) { + String key = args[i].substring(2); // 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; + } + + /** + * 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"); + } + } + + /** + * Validate the stream-results parameter value. + * + * @param streamResults the stream-results value to validate + * @throws IllegalArgumentException if the value is invalid + */ + static void validateStreamResults(String streamResults) { + if (streamResults == null || streamResults.isEmpty()) { + return; // stream-results is optional + } + + if (!"true".equalsIgnoreCase(streamResults) && !"false".equalsIgnoreCase(streamResults)) { + throw new IllegalArgumentException( + "Invalid stream-results value '" + streamResults + "'. Must be 'true' or 'false'"); + } + } + + /** + * 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 + */ + 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); + } + } + + /** 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 <bool> Whether to stream intermediate delete results\n" + + " Valid values: true, false\n" + + " --spark-conf <json> JSON map of custom Spark configurations\n" + + " Example: '{\"spark.sql.shuffle.partitions\":\"200\"}'\n" + + " Note: Cannot override catalog, extensions, or app name configs\n" + + "\n" Review Comment: The usage text says custom `--spark-conf` values “Cannot override catalog, extensions, or app name configs”, but the code applies every parsed key via `sparkBuilder.config(key, value)` with no filtering. This makes the usage output misleading; either enforce this (reject reserved keys) or reword it as an unsupported/should-not override warning. ########## maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergExpireSnapshotsJob.java: ########## @@ -0,0 +1,506 @@ +/* + * 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(12, template.arguments().size()); // 6 flags * 2 (flag + value) + + // Verify all expected arguments are present + assertTrue(template.arguments().contains("--catalog")); + assertTrue(template.arguments().contains("{{catalog_name}}")); + assertTrue(template.arguments().contains("--table")); + assertTrue(template.arguments().contains("{{table_identifier}}")); + assertTrue(template.arguments().contains("--older-than")); + assertTrue(template.arguments().contains("{{older_than}}")); + assertTrue(template.arguments().contains("--retain-last")); + assertTrue(template.arguments().contains("{{retain_last}}")); + assertTrue(template.arguments().contains("--stream-results")); + 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 = IcebergExpireSnapshotsJob.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 = IcebergExpireSnapshotsJob.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 = IcebergExpireSnapshotsJob.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 testParseArgumentsWithMissingValues() { + String[] args = {"--catalog", "iceberg_prod", "--table"}; + Map<String, String> result = IcebergExpireSnapshotsJob.parseArguments(args); + + // Only catalog should be parsed, table has no value + assertEquals(1, result.size()); + assertEquals("iceberg_prod", result.get("catalog")); + assertFalse(result.containsKey("table")); + } + + @Test + public void testParseArgumentsWithAllOptions() { + String[] args = { + "--catalog", "iceberg_prod", + "--table", "db.sample", + "--older-than", "2024-06-01 00:00:00", + "--retain-last", "3", + "--stream-results", "true", + "--spark-conf", "{\"spark.executor.memory\":\"4g\"}" + }; + Map<String, String> result = IcebergExpireSnapshotsJob.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 = IcebergExpireSnapshotsJob.parseArguments(args1); + Map<String, String> result2 = IcebergExpireSnapshotsJob.parseArguments(args2); + + assertEquals(result1, result2); + } + + // Test buildProcedureCall method + + @Test + public void testBuildProcedureCallMinimal() { + String sql = + IcebergExpireSnapshotsJob.buildProcedureCall("iceberg_prod", "db.sample", null, null, null); + + 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, null); + + 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", null); + + assertEquals( + "CALL iceberg_prod.system.expire_snapshots(table => 'db.sample', retain_last => 5)", sql); Review Comment: If the catalog identifier becomes backtick-quoted for safety, update this expected SQL to include the quoted catalog name. ########## maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergExpireSnapshotsJob.java: ########## @@ -0,0 +1,506 @@ +/* + * 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(12, template.arguments().size()); // 6 flags * 2 (flag + value) + + // Verify all expected arguments are present + assertTrue(template.arguments().contains("--catalog")); + assertTrue(template.arguments().contains("{{catalog_name}}")); + assertTrue(template.arguments().contains("--table")); + assertTrue(template.arguments().contains("{{table_identifier}}")); + assertTrue(template.arguments().contains("--older-than")); + assertTrue(template.arguments().contains("{{older_than}}")); + assertTrue(template.arguments().contains("--retain-last")); + assertTrue(template.arguments().contains("{{retain_last}}")); + assertTrue(template.arguments().contains("--stream-results")); + 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 = IcebergExpireSnapshotsJob.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 = IcebergExpireSnapshotsJob.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 = IcebergExpireSnapshotsJob.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 testParseArgumentsWithMissingValues() { + String[] args = {"--catalog", "iceberg_prod", "--table"}; + Map<String, String> result = IcebergExpireSnapshotsJob.parseArguments(args); + + // Only catalog should be parsed, table has no value + assertEquals(1, result.size()); + assertEquals("iceberg_prod", result.get("catalog")); + assertFalse(result.containsKey("table")); + } + + @Test + public void testParseArgumentsWithAllOptions() { + String[] args = { + "--catalog", "iceberg_prod", + "--table", "db.sample", + "--older-than", "2024-06-01 00:00:00", + "--retain-last", "3", + "--stream-results", "true", + "--spark-conf", "{\"spark.executor.memory\":\"4g\"}" + }; + Map<String, String> result = IcebergExpireSnapshotsJob.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 = IcebergExpireSnapshotsJob.parseArguments(args1); + Map<String, String> result2 = IcebergExpireSnapshotsJob.parseArguments(args2); + + assertEquals(result1, result2); + } + + // Test buildProcedureCall method + + @Test + public void testBuildProcedureCallMinimal() { + String sql = + IcebergExpireSnapshotsJob.buildProcedureCall("iceberg_prod", "db.sample", null, null, null); + + 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, null); + + 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", null); + + 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(")); Review Comment: If the catalog identifier becomes backtick-quoted for safety, this `startsWith` assertion should expect the quoted form. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
