diff --git a/apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/Operations.java b/apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/Operations.java index 6d15821aa..7e464930a 100644 --- a/apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/Operations.java +++ b/apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/Operations.java @@ -27,6 +27,7 @@ import java.util.concurrent.Executors; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; +import java.util.function.Consumer; import java.util.function.Predicate; import java.util.stream.Collectors; import lombok.AccessLevel; @@ -39,7 +40,6 @@ import org.apache.hadoop.fs.Path; import org.apache.hadoop.fs.RemoteIterator; import org.apache.iceberg.CatalogUtil; -import org.apache.iceberg.ExpireSnapshots; import org.apache.iceberg.FileScanTask; import org.apache.iceberg.Schema; import org.apache.iceberg.StructLike; @@ -47,6 +47,8 @@ import org.apache.iceberg.TableScan; import org.apache.iceberg.Transaction; import org.apache.iceberg.actions.DeleteOrphanFiles; +import org.apache.iceberg.actions.ExpireSnapshots; +import org.apache.iceberg.actions.ImmutableExpireSnapshots; import org.apache.iceberg.actions.RewriteDataFiles; import org.apache.iceberg.catalog.Catalog; import org.apache.iceberg.catalog.TableIdentifier; @@ -132,45 +134,76 @@ public DeleteOrphanFiles.Result deleteOrphanFiles( operation = operation.executeDeleteWith(removeFilesService(concurrentDeletes)); } Map dataManifestsCache = new ConcurrentHashMap<>(); - Path backupDirRoot = new Path(table.location(), backupDir); - Path dataDirRoot = new Path(table.location(), "data"); operation = operation.deleteWith( - file -> { - log.info("Detected orphan file {}", file); - if (file.endsWith("metadata.json")) { - // Don't remove metadata.json files since current metadata.json is recognized as - // orphan because of inclusion of the scheme in its file path returned by catalog. - // Also, we want Iceberg commits to remove the metadata.json files not the OFD job. - log.info("Skipped deleting metadata file {}", file); - } else if (file.contains(backupDirRoot.toString())) { - // files present in .backup dir should not be considered orphan - log.info("Skipped deleting backup file {}", file); - } else if (file.contains(dataDirRoot.toString()) - && isExistBackupDataManifests(table, file, backupDir, dataManifestsCache)) { - // move data files to backup dir when a data manifest exists for the partition, - // regardless of whether backup is currently enabled for the OFD job - Path backupFilePath = getTrashPath(table, file, backupDir); - log.info("Moving orphan file {} to {}", file, backupFilePath); - try { - rename(new Path(file), backupFilePath); - // update modification time to current time - fs().setTimes(backupFilePath, System.currentTimeMillis(), -1); - } catch (IOException e) { - log.error(String.format("Move operation failed for file: %s", file), e); - } - } else { - log.info("Deleting orphan file {}", file); - try { - fs().delete(new Path(file), false); - } catch (IOException e) { - log.error(String.format("Delete operation failed for file: %s", file), e); - } - } - }); + // OFD always moves orphan data files to the backup dir when a data manifest exists + // for the partition, regardless of whether backup is currently enabled for the OFD + // job, so backupEnabled is always passed as true here. + buildFileDeleteHandler(table, true, backupDir, dataManifestsCache)); return operation.execute(); } + /** + * Build a shared {@code deleteWith} handler used by both Orphan File Deletion (OFD) and Snapshot + * Expiration (SE) so the backup/delete semantics for expired or orphaned files stay in sync + * across both jobs rather than diverging as separate implementations. + * + *

Behavior: metadata.json files are skipped (Iceberg commits own their lifecycle), files + * already under the backup directory are skipped, data files are moved to the backup directory + * when backup is enabled and a data manifest backup already exists for that partition, otherwise + * the file is deleted directly. + */ + private Consumer buildFileDeleteHandler( + Table table, + boolean backupEnabled, + String backupDir, + Map dataManifestsCache) { + Path backupDirRoot = new Path(table.location(), backupDir); + Path dataDirRoot = new Path(table.location(), "data"); + return file -> { + log.info("Processing file for deletion {}", file); + if (file.endsWith("metadata.json")) { + // Don't remove metadata.json files since current metadata.json is recognized as + // orphan/expired because of inclusion of the scheme in its file path returned by catalog. + // Also, we want Iceberg commits to remove the metadata.json files not this job. + log.info("Skipped deleting metadata file {}", file); + } else if (file.contains(backupDirRoot.toString())) { + // files present in .backup dir should not be considered orphan/expired + log.info("Skipped deleting backup file {}", file); + } else if (file.contains(dataDirRoot.toString()) + && backupEnabled + && isExistBackupDataManifests(table, file, backupDir, dataManifestsCache)) { + // move data files to backup dir if backup is enabled + backupDataFile(file, table, backupDir); + } else { + deleteFile(file); + } + }; + } + + /** Move a data file to the backup directory. */ + private void backupDataFile(String file, Table table, String backupDir) { + Path backupFilePath = getTrashPath(table, file, backupDir); + log.info("Moving file {} to {}", file, backupFilePath); + try { + rename(new Path(file), backupFilePath); + // update modification time to current time + fs().setTimes(backupFilePath, System.currentTimeMillis(), -1); + } catch (IOException e) { + log.error(String.format("Move operation failed for file: %s", file), e); + } + } + + /** Delete a file directly (manifests, manifest lists, data files without a backup manifest). */ + private void deleteFile(String file) { + log.info("Deleting file {}", file); + try { + fs().delete(new Path(file), false); + } catch (IOException e) { + log.error(String.format("Delete operation failed for file: %s", file), e); + } + } + private ExecutorService removeFilesService(int concurrentDeletes) { return MoreExecutors.getExitingExecutorService( (ThreadPoolExecutor) @@ -262,7 +295,26 @@ public void deleteStagedOrphanDirectory( /** Expire snapshots on a given fully-qualified table name. */ public void expireSnapshots(String fqtn, int maxAge, String granularity, int versions) { - expireSnapshots(getTable(fqtn), maxAge, granularity, versions); + expireSnapshots(fqtn, maxAge, granularity, versions, false); + } + + /** Expire snapshots on a given fully-qualified table name with deleteFiles parameter. */ + public ExpireSnapshots.Result expireSnapshots( + String fqtn, int maxAge, String granularity, int versions, boolean deleteFiles) { + return expireSnapshots(fqtn, maxAge, granularity, versions, deleteFiles, false, ".backup"); + } + + /** Expire snapshots with backup support. */ + public ExpireSnapshots.Result expireSnapshots( + String fqtn, + int maxAge, + String granularity, + int versions, + boolean deleteFiles, + boolean backupEnabled, + String backupDir) { + return expireSnapshots( + getTable(fqtn), maxAge, granularity, versions, deleteFiles, backupEnabled, backupDir); } /** @@ -272,29 +324,164 @@ public void expireSnapshots(String fqtn, int maxAge, String granularity, int ver * number of snapshots younger than the maxAge */ public void expireSnapshots(Table table, int maxAge, String granularity, int versions) { - ExpireSnapshots expireSnapshotsCommand = table.expireSnapshots().cleanExpiredFiles(false); + // Call the Result-returning version but ignore the result for backward compatibility + expireSnapshots(table, maxAge, granularity, versions, false); + } + + /** + * Expire snapshots on a given {@link Table} with deleteFiles parameter. If maxAge is provided, it + * will expire snapshots older than maxAge in granularity timeunit. If versions is provided, it + * will retain the last versions snapshots. If both are provided, it will prioritize maxAge; only + * retain up to versions number of snapshots younger than the maxAge. Returns {@link + * ExpireSnapshots.Result} containing metrics about deleted files. + */ + public ExpireSnapshots.Result expireSnapshots( + Table table, int maxAge, String granularity, int versions, boolean deleteFiles) { + return expireSnapshots(table, maxAge, granularity, versions, deleteFiles, false, ".backup"); + } + + /** + * Expire snapshots with backup support. Main orchestration method that delegates to helper + * methods based on deleteFiles flag. + */ + public ExpireSnapshots.Result expireSnapshots( + Table table, + int maxAge, + String granularity, + int versions, + boolean deleteFiles, + boolean backupEnabled, + String backupDir) { - // maxAge will always be defined ChronoUnit timeUnitGranularity = ChronoUnit.valueOf( SparkJobUtil.convertGranularityToChrono(granularity.toUpperCase()).name()); long expireBeforeTimestampMs = System.currentTimeMillis() - timeUnitGranularity.getDuration().multipliedBy(maxAge).toMillis(); - log.info("Expiring snapshots for table: {} older than {}ms", table, expireBeforeTimestampMs); - expireSnapshotsCommand.expireOlderThan(expireBeforeTimestampMs).commit(); + log.info( + "Expiring snapshots for table: {} older than {}ms with deleteFiles={}, backupEnabled={}, backupDir={}", + table, + expireBeforeTimestampMs, + deleteFiles, + backupEnabled, + backupDir); + + // First expiration: based on maxAge + ExpireSnapshots.Result result = + deleteFiles + ? expireSnapshotsWithFiles( + table, expireBeforeTimestampMs, null, backupEnabled, backupDir) + : expireSnapshotsMetadataOnly(table, expireBeforeTimestampMs, null); + + // Second expiration: based on versions (if needed). Both phases can delete files, so their + // results are combined rather than the second phase's result silently replacing the first's + // and dropping its deleted-file/manifest counts from observability (metrics and logs). if (versions > 0 && Iterators.size(table.snapshots().iterator()) > versions) { log.info("Expiring snapshots for table: {} retaining last {} versions", table, versions); - // Note: retainLast keeps the last N snapshots that WOULD be expired, hence expireOlderThan - // currentTime - expireSnapshotsCommand - .expireOlderThan(System.currentTimeMillis()) - .retainLast(versions) - .commit(); + ExpireSnapshots.Result versionsResult = + deleteFiles + ? expireSnapshotsWithFiles( + table, System.currentTimeMillis(), versions, backupEnabled, backupDir) + : expireSnapshotsMetadataOnly(table, System.currentTimeMillis(), versions); + result = combineResults(result, versionsResult); + } + + return result; + } + + /** + * Combine two {@link ExpireSnapshots.Result} instances by summing their deleted-file/manifest + * counts. Used to accumulate results across the maxAge-based and versions-based expiration phases + * so metrics/logs reflect files deleted by both phases instead of only the last one run. + */ + @VisibleForTesting + static ExpireSnapshots.Result combineResults( + ExpireSnapshots.Result first, ExpireSnapshots.Result second) { + return ImmutableExpireSnapshots.Result.builder() + .deletedDataFilesCount(first.deletedDataFilesCount() + second.deletedDataFilesCount()) + .deletedManifestsCount(first.deletedManifestsCount() + second.deletedManifestsCount()) + .deletedManifestListsCount( + first.deletedManifestListsCount() + second.deletedManifestListsCount()) + .deletedPositionDeleteFilesCount( + first.deletedPositionDeleteFilesCount() + second.deletedPositionDeleteFilesCount()) + .deletedEqualityDeleteFilesCount( + first.deletedEqualityDeleteFilesCount() + second.deletedEqualityDeleteFilesCount()) + .build(); + } + + /** + * Expire snapshots using Table API (metadata-only, no file deletion). Efficient - skips file + * enumeration entirely. + */ + private ExpireSnapshots.Result expireSnapshotsMetadataOnly( + Table table, long expireBeforeTimestampMs, Integer versions) { + + org.apache.iceberg.ExpireSnapshots expireSnapshots = + table.expireSnapshots().cleanExpiredFiles(false).expireOlderThan(expireBeforeTimestampMs); + + if (versions != null) { + expireSnapshots = expireSnapshots.retainLast(versions); + } + + expireSnapshots.commit(); + + // Return empty result for metadata-only operation + return ImmutableExpireSnapshots.Result.builder() + .deletedDataFilesCount(0L) + .deletedManifestsCount(0L) + .deletedManifestListsCount(0L) + .deletedPositionDeleteFilesCount(0L) + .deletedEqualityDeleteFilesCount(0L) + .build(); + } + + /** + * Expire snapshots using SparkActions API with file deletion. Snapshot planning/manifest + * enumeration is distributed via Spark, with optional backup support; the delete/backup callback + * itself runs driver-side (see {@link #createExpireSnapshotsActionWithBackup}). + */ + private ExpireSnapshots.Result expireSnapshotsWithFiles( + Table table, + long expireBeforeTimestampMs, + Integer versions, + boolean backupEnabled, + String backupDir) { + + ExpireSnapshots expireSnapshotsAction = + backupEnabled + ? createExpireSnapshotsActionWithBackup(table, backupDir) + : SparkActions.get(spark).expireSnapshots(table); + + expireSnapshotsAction = expireSnapshotsAction.expireOlderThan(expireBeforeTimestampMs); + + if (versions != null) { + expireSnapshotsAction = expireSnapshotsAction.retainLast(versions); } + + return expireSnapshotsAction.execute(); } + /** + * Create ExpireSnapshots action with custom backup logic for data files. The {@code deleteWith} + * callback (backup-or-delete decision) runs driver-side via Iceberg's internal executor pool, not + * distributed across Spark executors; only manifest/snapshot planning is distributed. + */ + private ExpireSnapshots createExpireSnapshotsActionWithBackup(Table table, String backupDir) { + Map dataManifestsCache = new ConcurrentHashMap<>(); + + return SparkActions.get(spark) + .expireSnapshots(table) + .deleteWith(buildFileDeleteHandler(table, true, backupDir, dataManifestsCache)); + } + + /* + * NOTE: file backup/delete semantics (metadata.json skip, backup-dir skip, backup-or-delete + * for data files) are shared with Orphan File Deletion via buildFileDeleteHandler(...) above, + * so both jobs stay consistent instead of maintaining separate implementations. + */ + /** * Run table retention operation if there are rows with partition column (@columnName) value older * than @count @granularity. diff --git a/apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/SnapshotsExpirationSparkApp.java b/apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/SnapshotsExpirationSparkApp.java index 36c40efd3..d5273580a 100644 --- a/apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/SnapshotsExpirationSparkApp.java +++ b/apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/SnapshotsExpirationSparkApp.java @@ -3,6 +3,7 @@ import com.linkedin.openhouse.common.metrics.DefaultOtelConfig; import com.linkedin.openhouse.common.metrics.OtelEmitter; import com.linkedin.openhouse.jobs.spark.state.StateManager; +import com.linkedin.openhouse.jobs.util.AppConstants; import com.linkedin.openhouse.jobs.util.AppsOtelEmitter; import java.time.temporal.ChronoUnit; import java.util.ArrayList; @@ -11,6 +12,8 @@ import lombok.extern.slf4j.Slf4j; import org.apache.commons.cli.CommandLine; import org.apache.commons.cli.Option; +import org.apache.iceberg.Table; +import org.apache.iceberg.actions.ExpireSnapshots; /** * Class with main entry point to run as a table snapshot expiration job. Snapshots for table which @@ -24,11 +27,14 @@ public class SnapshotsExpirationSparkApp extends BaseTableSparkApp { private final String granularity; private final int maxAge; private final int versions; + private final boolean deleteFiles; + private final String backupDir; public static class DEFAULT_CONFIGURATION { public static final int MAX_AGE = 3; public static final String GRANULARITY = ChronoUnit.DAYS.toString(); public static final int VERSIONS = 0; + public static final boolean DELETE_FILES = false; } public SnapshotsExpirationSparkApp( @@ -38,6 +44,8 @@ public SnapshotsExpirationSparkApp( int maxAge, String granularity, int versions, + boolean deleteFiles, + String backupDir, OtelEmitter otelEmitter) { super(jobId, stateManager, fqtn, otelEmitter); // By default, always enforce a time to live for snapshots even if unconfigured @@ -49,17 +57,59 @@ public SnapshotsExpirationSparkApp( this.granularity = granularity; } this.versions = versions; + this.deleteFiles = deleteFiles; + this.backupDir = backupDir; } @Override protected void runInner(Operations ops) { + Table table = ops.getTable(fqtn); + boolean backupEnabled = + Boolean.parseBoolean( + table.properties().getOrDefault(AppConstants.BACKUP_ENABLED_KEY, "false")); + log.info( - "Snapshot expiration app start for table {}, expiring older than {} {}s or with more than {} versions", + "Snapshot expiration app start for table {}, expiring older than {} {}s or with more than {} versions, deleteFiles={}, backupEnabled={}, backupDir={}", fqtn, maxAge, granularity, - versions); - ops.expireSnapshots(fqtn, maxAge, granularity, versions); + versions, + deleteFiles, + backupEnabled, + backupDir); + + long startTime = System.currentTimeMillis(); + ExpireSnapshots.Result result = + ops.expireSnapshots( + fqtn, maxAge, granularity, versions, deleteFiles, backupEnabled, backupDir); + long duration = System.currentTimeMillis() - startTime; + + // Log results + log.info( + "Snapshot expiration completed for table {}. Deleted {} data files, {} equality delete files, {} position delete files, {} manifests, {} manifest lists", + fqtn, + result.deletedDataFilesCount(), + result.deletedEqualityDeleteFilesCount(), + result.deletedPositionDeleteFilesCount(), + result.deletedManifestsCount(), + result.deletedManifestListsCount()); + + // Emit metrics + recordMetrics(duration); + } + + private void recordMetrics(long duration) { + io.opentelemetry.api.common.Attributes attributes = + io.opentelemetry.api.common.Attributes.of( + io.opentelemetry.api.common.AttributeKey.stringKey(AppConstants.TABLE_NAME), + fqtn, + io.opentelemetry.api.common.AttributeKey.booleanKey(AppConstants.DELETE_FILES_ENABLED), + deleteFiles); + otelEmitter.time( + SnapshotsExpirationSparkApp.class.getName(), + AppConstants.SNAPSHOTS_EXPIRATION_DURATION, + duration, + attributes); } public static void main(String[] args) { @@ -76,6 +126,14 @@ public static SnapshotsExpirationSparkApp createApp(String[] args, OtelEmitter o extraOptions.add(new Option("g", "granularity", true, "Granularity: day")); extraOptions.add( new Option("v", "versions", true, "Number of versions to keep after snapshot expiration")); + extraOptions.add( + new Option( + "d", + "deleteFiles", + false, + "Delete expired snapshot files (data, manifests, manifest lists)")); + extraOptions.add( + new Option("b", "backupDir", true, "Backup directory for data files (default: .backup)")); CommandLine cmdLine = createCommandLine(args, extraOptions); return new SnapshotsExpirationSparkApp( getJobId(cmdLine), @@ -84,6 +142,8 @@ public static SnapshotsExpirationSparkApp createApp(String[] args, OtelEmitter o Integer.parseInt(cmdLine.getOptionValue("maxAge", "0")), cmdLine.getOptionValue("granularity", ""), Integer.parseInt(cmdLine.getOptionValue("versions", "0")), + cmdLine.hasOption("deleteFiles"), + cmdLine.getOptionValue("backupDir", ".backup"), otelEmitter); } } diff --git a/apps/spark/src/main/java/com/linkedin/openhouse/jobs/util/AppConstants.java b/apps/spark/src/main/java/com/linkedin/openhouse/jobs/util/AppConstants.java index bcafacc53..a9824ff3b 100644 --- a/apps/spark/src/main/java/com/linkedin/openhouse/jobs/util/AppConstants.java +++ b/apps/spark/src/main/java/com/linkedin/openhouse/jobs/util/AppConstants.java @@ -23,6 +23,10 @@ public final class AppConstants { public static final String RETENTION_POLICY_MISCONFIGURED_TABLE_COUNT = "retention_policy_misconfigured_table_count"; + // Snapshot Expiration metrics + public static final String SNAPSHOTS_EXPIRATION_DURATION = "snapshots_expiration_duration"; + public static final String DELETE_FILES_ENABLED = "delete_files_enabled"; + // Openhouse jobs status tags public static final String STATUS = "status"; public static final String STATUS_CODE = "status_code"; diff --git a/apps/spark/src/test/java/com/linkedin/openhouse/jobs/scheduler/tasks/SnapshotExpirationTaskTest.java b/apps/spark/src/test/java/com/linkedin/openhouse/jobs/scheduler/tasks/SnapshotExpirationTaskTest.java index bc284bb46..3da5cfc42 100644 --- a/apps/spark/src/test/java/com/linkedin/openhouse/jobs/scheduler/tasks/SnapshotExpirationTaskTest.java +++ b/apps/spark/src/test/java/com/linkedin/openhouse/jobs/scheduler/tasks/SnapshotExpirationTaskTest.java @@ -104,4 +104,113 @@ void testSnapshotExpirationJobWithMaxAgeAndVersions() { .collect(Collectors.toList()); Assertions.assertEquals(expectedArgs, tableRetentionTask.getArgs()); } + + @Test + void testSnapshotExpirationWithDeleteFilesDisabled() { + TableSnapshotsExpirationTask tableRetentionTask = + new TableSnapshotsExpirationTask(jobsClient, tablesClient, tableMetadata); + + List expectedArgs = + Stream.of("--tableName", tableMetadata.fqtn()).collect(Collectors.toList()); + Assertions.assertEquals(expectedArgs, tableRetentionTask.getArgs()); + } + + @Test + void testSnapshotExpirationWithDeleteFilesEnabled() { + TableSnapshotsExpirationTask tableRetentionTask = + new TableSnapshotsExpirationTask(jobsClient, tablesClient, tableMetadata, 1000, 2000, 3000); + + // deleteFiles is no longer passed via scheduler, so it won't be in args + List expectedArgs = + Stream.of("--tableName", tableMetadata.fqtn()).collect(Collectors.toList()); + Assertions.assertEquals(expectedArgs, tableRetentionTask.getArgs()); + } + + @Test + void testSnapshotExpirationWithDeleteFilesAndMaxAgeConfig() { + TableSnapshotsExpirationTask tableRetentionTask = + new TableSnapshotsExpirationTask(jobsClient, tablesClient, tableMetadata, 1000, 2000, 3000); + + HistoryConfig historyConfigMock = Mockito.mock(HistoryConfig.class); + int maxAge = 1; + History.GranularityEnum granularity = History.GranularityEnum.DAY; + + Mockito.when(tableMetadata.getHistoryConfig()).thenReturn(historyConfigMock); + Mockito.when(historyConfigMock.getMaxAge()).thenReturn(maxAge); + Mockito.when(historyConfigMock.getGranularity()).thenReturn(granularity); + List expectedArgs = + Stream.of( + "--tableName", + tableMetadata.fqtn(), + "--maxAge", + String.valueOf(maxAge), + "--granularity", + granularity.getValue()) + .collect(Collectors.toList()); + Assertions.assertEquals(expectedArgs, tableRetentionTask.getArgs()); + } + + @Test + void testSnapshotExpirationWithDeleteFilesAndVersionsConfig() { + TableSnapshotsExpirationTask tableRetentionTask = + new TableSnapshotsExpirationTask(jobsClient, tablesClient, tableMetadata, 1000, 2000, 3000); + + HistoryConfig historyConfigMock = Mockito.mock(HistoryConfig.class); + int versions = 3; + + Mockito.when(tableMetadata.getHistoryConfig()).thenReturn(historyConfigMock); + Mockito.when(historyConfigMock.getVersions()).thenReturn(versions); + List expectedArgs = + Stream.of("--tableName", tableMetadata.fqtn(), "--versions", String.valueOf(versions)) + .collect(Collectors.toList()); + Assertions.assertEquals(expectedArgs, tableRetentionTask.getArgs()); + } + + @Test + void testSnapshotExpirationWithDeleteFilesAndFullConfig() { + TableSnapshotsExpirationTask tableRetentionTask = + new TableSnapshotsExpirationTask(jobsClient, tablesClient, tableMetadata, 1000, 2000, 3000); + + HistoryConfig historyConfigMock = Mockito.mock(HistoryConfig.class); + int maxAge = 3; + History.GranularityEnum granularity = History.GranularityEnum.DAY; + int versions = 3; + + Mockito.when(tableMetadata.getHistoryConfig()).thenReturn(historyConfigMock); + Mockito.when(historyConfigMock.getMaxAge()).thenReturn(maxAge); + Mockito.when(historyConfigMock.getGranularity()).thenReturn(granularity); + Mockito.when(historyConfigMock.getVersions()).thenReturn(versions); + + List expectedArgs = + Stream.of( + "--tableName", + tableMetadata.fqtn(), + "--maxAge", + String.valueOf(maxAge), + "--granularity", + granularity.getValue(), + "--versions", + String.valueOf(versions)) + .collect(Collectors.toList()); + Assertions.assertEquals(expectedArgs, tableRetentionTask.getArgs()); + } + + @Test + void testSnapshotExpirationWithTimeoutsAndDeleteFiles() { + long pollIntervalMs = 1000L; + long queuedTimeoutMs = 5000L; + long taskTimeoutMs = 10000L; + TableSnapshotsExpirationTask tableRetentionTask = + new TableSnapshotsExpirationTask( + jobsClient, + tablesClient, + tableMetadata, + pollIntervalMs, + queuedTimeoutMs, + taskTimeoutMs); + + List expectedArgs = + Stream.of("--tableName", tableMetadata.fqtn()).collect(Collectors.toList()); + Assertions.assertEquals(expectedArgs, tableRetentionTask.getArgs()); + } } diff --git a/apps/spark/src/test/java/com/linkedin/openhouse/jobs/spark/OperationsTest.java b/apps/spark/src/test/java/com/linkedin/openhouse/jobs/spark/OperationsTest.java index db8ff7e98..c1f56b37a 100644 --- a/apps/spark/src/test/java/com/linkedin/openhouse/jobs/spark/OperationsTest.java +++ b/apps/spark/src/test/java/com/linkedin/openhouse/jobs/spark/OperationsTest.java @@ -32,6 +32,7 @@ import java.util.stream.Collectors; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.tuple.Triple; +import org.apache.hadoop.fs.FileStatus; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.iceberg.Schema; @@ -39,6 +40,8 @@ import org.apache.iceberg.Table; import org.apache.iceberg.Transaction; import org.apache.iceberg.actions.DeleteOrphanFiles; +import org.apache.iceberg.actions.ExpireSnapshots; +import org.apache.iceberg.actions.ImmutableExpireSnapshots; import org.apache.iceberg.actions.RewriteDataFiles; import org.apache.iceberg.types.Types; import org.apache.spark.sql.Row; @@ -600,6 +603,38 @@ public void testSnapshotsExpirationVersionsNoop() throws Exception { } } + @Test + public void testCombineResultsSumsCountsAcrossPhases() { + // Regression test for a bug where Operations#expireSnapshots overwrote the maxAge-phase + // Result with the versions-phase Result instead of accumulating them, silently dropping the + // first phase's deleted file/manifest counts from metrics and logs. + ExpireSnapshots.Result maxAgePhaseResult = + ImmutableExpireSnapshots.Result.builder() + .deletedDataFilesCount(2L) + .deletedManifestsCount(3L) + .deletedManifestListsCount(1L) + .deletedPositionDeleteFilesCount(4L) + .deletedEqualityDeleteFilesCount(5L) + .build(); + ExpireSnapshots.Result versionsPhaseResult = + ImmutableExpireSnapshots.Result.builder() + .deletedDataFilesCount(10L) + .deletedManifestsCount(20L) + .deletedManifestListsCount(30L) + .deletedPositionDeleteFilesCount(40L) + .deletedEqualityDeleteFilesCount(50L) + .build(); + + ExpireSnapshots.Result combined = + Operations.combineResults(maxAgePhaseResult, versionsPhaseResult); + + Assertions.assertEquals(12L, combined.deletedDataFilesCount()); + Assertions.assertEquals(23L, combined.deletedManifestsCount()); + Assertions.assertEquals(31L, combined.deletedManifestListsCount()); + Assertions.assertEquals(44L, combined.deletedPositionDeleteFilesCount()); + Assertions.assertEquals(55L, combined.deletedEqualityDeleteFilesCount()); + } + @Test public void testSnapshotsExpirationVersions() throws Exception { final String tableName = "db.test_es_versions_java"; @@ -668,6 +703,72 @@ public void testSnapshotsExpirationBothAgeAndVersions() throws Exception { } } + @Test + public void testSnapshotsExpirationAccumulatesResultsAcrossBothPhases() throws Exception { + // Regression test for a bug where, when both the maxAge-based and versions-based expiration + // phases run and delete files, the versions-phase Result silently replaced (instead of being + // combined with) the maxAge-phase Result, dropping the first phase's deleted-file counts from + // the returned Result (and therefore from SnapshotsExpirationSparkApp's logs/metrics). + // + // Setup: an old snapshot (S1) is created, its data file is orphaned via RTAS (S2), then a + // recent append (S3) is added. maxAge expires only S1 (deleting its now-orphaned data file); + // versions (retain last 1) then further expires S2. Only the maxAge phase deletes a data + // file; if its contribution were dropped, the final result would incorrectly show 0. + final String tableName = "db.test_es_accumulate_results"; + final String sourceName = "db.test_es_accumulate_results_source"; + final int maxAge = 20; + final String timeGranularity = "SECONDS"; + final int versionsToKeep = 1; + + List snapshotIds; + try (Operations ops = Operations.withCatalog(getSparkSession(), otelEmitter)) { + prepareTable(ops, tableName); + populateTable(ops, tableName, 1); + Thread.sleep(30000); // Sleep to age the first snapshot past maxAge + + ops.spark() + .sql( + String.format( + "CREATE TABLE %s (data string, ts timestamp) USING iceberg", sourceName)); + ops.spark() + .sql(String.format("INSERT INTO %s VALUES ('a', current_timestamp())", sourceName)); + ops.spark() + .sql( + String.format( + "ALTER TABLE %s SET TBLPROPERTIES ('replace.enabled'='true')", tableName)); + // RTAS orphans the original (now-aged) data file from the first insert. + ops.spark() + .sql( + String.format( + "REPLACE TABLE %s USING iceberg AS SELECT * FROM %s", tableName, sourceName)); + populateTable(ops, tableName, 1); // recent append on top of the RTAS snapshot + + Table table = ops.getTable(tableName); + snapshotIds = getSnapshotIds(ops, tableName); + Assertions.assertEquals( + 3, snapshotIds.size(), "Should have 3 snapshots: original insert, RTAS, and append"); + + ExpireSnapshots.Result result = + ops.expireSnapshots(table, maxAge, timeGranularity, versionsToKeep, true); + Assertions.assertNotNull(result, "Result should not be null"); + + // Only retain the last (most recent append) snapshot. + checkSnapshots(table, snapshotIds.subList(snapshotIds.size() - 1, snapshotIds.size())); + + Assertions.assertTrue( + result.deletedDataFilesCount() > 0, + "The maxAge phase's deleted data file (orphaned by RTAS) must be reflected in the " + + "combined result, not dropped by the subsequent versions phase"); + } + + // restart the app to reload catalog cache + try (Operations ops = Operations.withCatalog(getSparkSession(), otelEmitter)) { + checkSnapshots( + ops, tableName, snapshotIds.subList(snapshotIds.size() - 1, snapshotIds.size())); + ops.spark().sql(String.format("DROP TABLE IF EXISTS %s", sourceName)); + } + } + @Test public void testSnapshotsExpirationPrioritizeAge() throws Exception { final String tableName = "db.test_es_age_prioritization_java"; @@ -796,6 +897,254 @@ public void testSnapshotsExpirationAfterReplaceTable() throws Exception { } } + @Test + public void testSnapshotsExpirationWithFilesDeletion() throws Exception { + final String tableName = "db.test_es_delete_files"; + final int numInserts = 5; + final int maxAge = 0; + final String timeGranularity = "DAYS"; + + List snapshotIds; + try (Operations ops = Operations.withCatalog(getSparkSession(), otelEmitter)) { + prepareTable(ops, tableName); + populateTable(ops, tableName, numInserts); + snapshotIds = getSnapshotIds(ops, tableName); + Assertions.assertEquals( + numInserts, + snapshotIds.size(), + String.format("There must be %d snapshot(s) after inserts", numInserts)); + Table table = ops.getTable(tableName); + log.info("Loaded table {}, location {}", table.name(), table.location()); + + // Expire snapshots with deleteFiles=true + org.apache.iceberg.actions.ExpireSnapshots.Result resultWithDeletion = + ops.expireSnapshots(table, maxAge, timeGranularity, 0, true); + + // Verify that the result object is returned properly + log.info( + "Snapshot expiration with deleteFiles=true: deleted {} data files, {} equality delete files, {} position delete files, {} manifests, {} manifest lists", + resultWithDeletion.deletedDataFilesCount(), + resultWithDeletion.deletedEqualityDeleteFilesCount(), + resultWithDeletion.deletedPositionDeleteFilesCount(), + resultWithDeletion.deletedManifestsCount(), + resultWithDeletion.deletedManifestListsCount()); + + // Verify result is not null and has the expected structure + Assertions.assertNotNull(resultWithDeletion, "Result should not be null"); + + // When deleteFiles=true, manifests and manifest lists should be deleted + // Data files may or may not be deleted depending on whether they're still referenced + Assertions.assertTrue( + resultWithDeletion.deletedManifestsCount() > 0 + || resultWithDeletion.deletedManifestListsCount() > 0, + "Should have deleted manifests or manifest lists from expired snapshots"); + + // Only retain the last snapshot + checkSnapshots(table, snapshotIds.subList(snapshotIds.size() - 1, snapshotIds.size())); + } + + // restart the app to reload catalog cache + try (Operations ops = Operations.withCatalog(getSparkSession(), otelEmitter)) { + // verify that new apps see snapshots correctly + checkSnapshots( + ops, tableName, snapshotIds.subList(snapshotIds.size() - 1, snapshotIds.size())); + } + } + + @Test + public void testSnapshotsExpirationWithoutFilesDeletion() throws Exception { + final String tableName = "db.test_es_no_delete_files"; + final int numInserts = 5; + final int maxAge = 0; + final String timeGranularity = "DAYS"; + + List snapshotIds; + try (Operations ops = Operations.withCatalog(getSparkSession(), otelEmitter)) { + prepareTable(ops, tableName); + populateTable(ops, tableName, numInserts); + snapshotIds = getSnapshotIds(ops, tableName); + Assertions.assertEquals( + numInserts, + snapshotIds.size(), + String.format("There must be %d snapshot(s) after inserts", numInserts)); + Table table = ops.getTable(tableName); + log.info("Loaded table {}, location {}", table.name(), table.location()); + + // Expire snapshots with deleteFiles=false (default behavior) + org.apache.iceberg.actions.ExpireSnapshots.Result resultWithoutDeletion = + ops.expireSnapshots(table, maxAge, timeGranularity, 0, false); + + // Verify that no files were deleted (custom delete function prevents deletion) + log.info( + "Snapshot expiration with deleteFiles=false: deleted {} data files, {} manifests, {} manifest lists", + resultWithoutDeletion.deletedDataFilesCount(), + resultWithoutDeletion.deletedManifestsCount(), + resultWithoutDeletion.deletedManifestListsCount()); + + // With deleteFiles=false, no files should be physically deleted + Assertions.assertEquals( + 0, + resultWithoutDeletion.deletedDataFilesCount(), + "Should not delete data files when deleteFiles=false"); + Assertions.assertEquals( + 0, + resultWithoutDeletion.deletedManifestsCount(), + "Should not delete manifest files when deleteFiles=false"); + + // Only retain the last snapshot + checkSnapshots(table, snapshotIds.subList(snapshotIds.size() - 1, snapshotIds.size())); + } + + // restart the app to reload catalog cache + try (Operations ops = Operations.withCatalog(getSparkSession(), otelEmitter)) { + // verify that new apps see snapshots correctly + checkSnapshots( + ops, tableName, snapshotIds.subList(snapshotIds.size() - 1, snapshotIds.size())); + } + } + + // The following two tests exercise Operations#buildFileDeleteHandler, the delete-handling logic + // shared between Snapshot Expiration (SE) and Orphan File Deletion (OFD), from the SE side. + // This ensures both jobs stay consistent instead of maintaining separate implementations. + // Table content is replaced via RTAS (rather than plain appends) so that the original data + // files actually become unreferenced by the current snapshot and are therefore eligible for + // physical deletion/backup once their originating snapshots expire. + @Test + public void testSnapshotsExpirationWithBackupMovesDataFilesToBackupDir() throws Exception { + final String tableName = "db.test_es_backup_delete_files"; + final String sourceName = "db.test_es_backup_delete_files_source"; + final int numInserts = 3; + final int maxAge = 0; + final String timeGranularity = "DAYS"; + + List snapshotIds; + try (Operations ops = Operations.withCatalog(getSparkSession(), otelEmitter)) { + ops.spark() + .sql( + String.format( + "CREATE TABLE %s (data string, ts timestamp) USING iceberg", sourceName)); + ops.spark() + .sql( + String.format( + "INSERT INTO %s VALUES ('a', current_timestamp()), ('b', current_timestamp())", + sourceName)); + + prepareTable(ops, tableName); + populateTable(ops, tableName, numInserts); + + // RTAS is disabled by default; opt the table in before replacing it. + ops.spark() + .sql( + String.format( + "ALTER TABLE %s SET TBLPROPERTIES ('replace.enabled'='true')", tableName)); + // replace the table so original data files become unreferenced by the current snapshot + ops.spark() + .sql( + String.format( + "REPLACE TABLE %s USING iceberg AS SELECT * FROM %s", tableName, sourceName)); + + Table table = ops.getTable(tableName); + snapshotIds = getSnapshotIds(ops, tableName); + Assertions.assertTrue( + snapshotIds.size() > 1, "Should have multiple snapshots after inserts and RTAS"); + FileSystem fs = ops.fs(); + + // Simulate an existing backup manifest for the data partition so the shared delete handler + // treats expired data files as already backed-up (mirrors OFD's backup-manifest check) and + // moves them to the backup directory instead of deleting them outright. + Path dataManifestPath = + new Path(table.location(), BACKUP_DIR + "/data/data_manifest_pre.json"); + fs.createNewFile(dataManifestPath); + + org.apache.iceberg.actions.ExpireSnapshots.Result result = + ops.expireSnapshots(table, maxAge, timeGranularity, 0, true, true, BACKUP_DIR); + Assertions.assertNotNull(result, "Result should not be null"); + + // Only retain the last snapshot (the RTAS one) + checkSnapshots(table, snapshotIds.subList(snapshotIds.size() - 1, snapshotIds.size())); + + // The original (pre-RTAS) data files should have been moved to the backup directory rather + // than deleted, since a data manifest already exists for their backup partition. + FileStatus[] backedUpDataFiles = + fs.globStatus(new Path(table.location(), BACKUP_DIR + "/data/*.orc")); + Assertions.assertNotNull(backedUpDataFiles); + Assertions.assertTrue( + backedUpDataFiles.length > 0, + "Expired data files should be backed up when a data manifest already exists for the" + + " partition"); + } + + // restart the app to reload catalog cache + try (Operations ops = Operations.withCatalog(getSparkSession(), otelEmitter)) { + checkSnapshots( + ops, tableName, snapshotIds.subList(snapshotIds.size() - 1, snapshotIds.size())); + ops.spark().sql(String.format("DROP TABLE IF EXISTS %s", sourceName)); + } + } + + @Test + public void testSnapshotsExpirationWithoutBackupDeletesDataFilesDirectly() throws Exception { + final String tableName = "db.test_es_no_backup_delete_files"; + final String sourceName = "db.test_es_no_backup_delete_files_source"; + final int numInserts = 3; + final int maxAge = 0; + final String timeGranularity = "DAYS"; + + List snapshotIds; + try (Operations ops = Operations.withCatalog(getSparkSession(), otelEmitter)) { + ops.spark() + .sql( + String.format( + "CREATE TABLE %s (data string, ts timestamp) USING iceberg", sourceName)); + ops.spark() + .sql( + String.format( + "INSERT INTO %s VALUES ('a', current_timestamp()), ('b', current_timestamp())", + sourceName)); + + prepareTable(ops, tableName); + populateTable(ops, tableName, numInserts); + + ops.spark() + .sql( + String.format( + "ALTER TABLE %s SET TBLPROPERTIES ('replace.enabled'='true')", tableName)); + ops.spark() + .sql( + String.format( + "REPLACE TABLE %s USING iceberg AS SELECT * FROM %s", tableName, sourceName)); + + Table table = ops.getTable(tableName); + snapshotIds = getSnapshotIds(ops, tableName); + Assertions.assertTrue( + snapshotIds.size() > 1, "Should have multiple snapshots after inserts and RTAS"); + FileSystem fs = ops.fs(); + + // A data manifest is present, but backup is disabled: the shared delete handler must still + // delete the expired data files directly rather than moving them to the backup directory. + Path dataManifestPath = + new Path(table.location(), BACKUP_DIR + "/data/data_manifest_pre.json"); + fs.createNewFile(dataManifestPath); + + ops.expireSnapshots(table, maxAge, timeGranularity, 0, true, false, BACKUP_DIR); + + checkSnapshots(table, snapshotIds.subList(snapshotIds.size() - 1, snapshotIds.size())); + + FileStatus[] backedUpDataFiles = + fs.globStatus(new Path(table.location(), BACKUP_DIR + "/data/*.orc")); + Assertions.assertTrue( + backedUpDataFiles == null || backedUpDataFiles.length == 0, + "No data files should be backed up when backupEnabled=false"); + } + + // restart the app to reload catalog cache + try (Operations ops = Operations.withCatalog(getSparkSession(), otelEmitter)) { + checkSnapshots( + ops, tableName, snapshotIds.subList(snapshotIds.size() - 1, snapshotIds.size())); + ops.spark().sql(String.format("DROP TABLE IF EXISTS %s", sourceName)); + } + } + @Test public void testStagedFilesDelete() throws Exception { final String tableName = "db.test_staged_delete"; diff --git a/infra/recipes/docker-compose/oh-hadoop-spark/jobs.yaml b/infra/recipes/docker-compose/oh-hadoop-spark/jobs.yaml index 58cce65ce..d6a139339 100644 --- a/infra/recipes/docker-compose/oh-hadoop-spark/jobs.yaml +++ b/infra/recipes/docker-compose/oh-hadoop-spark/jobs.yaml @@ -47,7 +47,7 @@ jobs: << : *livy-engine - type: SNAPSHOTS_EXPIRATION class-name: com.linkedin.openhouse.jobs.spark.SnapshotsExpirationSparkApp - args: [] + args: ["--deleteFiles", "--backupDir", ".backup"] << : *apps-defaults << : *livy-engine - type: ORPHAN_FILES_DELETION