Make snapshots expiration job leaner (#131)

## Summary
Snapshots expiration job skips removing files as we wanted to localize
files removal to one job. The job traverses the files tree though, and
that is expensive and unnecessary. As the result of this change, it
updates snapshots list in the metadata without traversing the files
tree.

## Changes

- [ ] Client-facing API Changes
- [ ] Internal API Changes
- [ ] Bug Fixes
- [ ] New Features
- [x] Performance Improvements
- [ ] Code Style
- [ ] Refactoring
- [ ] Documentation
- [ ] Tests

For all the boxes checked, please include additional details of the
changes made in this pull request.

## Testing Done
<!--- Check any relevant boxes with "x" -->

- [ ] Manually Tested on local docker setup. Please include commands
ran, and their output.
- [ ] Added new tests for the changes made.
- [ ] Updated existing tests to reflect the changes made.
- [x] No tests added or updated. Please explain why. If unsure, please
feel free to ask for help.
- [ ] Some other form of testing like staging or soak time in
production. Please explain.

No change in the job effect is expected, it's purely optimization.

For all the boxes checked, include a detailed description of the testing
done for the changes made in this pull request.

# Additional Information

- [ ] Breaking Changes
- [ ] Deprecations
- [ ] Large PR broken into smaller PRs, and PR plan linked in the
description.

For all the boxes checked, include additional details of the changes
made in this pull request.

apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/Operations.java

@@ -26,7 +26,6 @@
 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.RewriteDataFiles;
 import org.apache.iceberg.catalog.Catalog;
 import org.apache.iceberg.catalog.TableIdentifier;
@@ -201,22 +200,18 @@ public void deleteStagedOrphanDirectory(
     }
   }
 
-  /** Run ExpireSnapshots operation on a given fully-qualified table name. */
-  public ExpireSnapshots.Result expireSnapshots(String fqtn, long expireBeforeTimestampMs) {
-    return expireSnapshots(getTable(fqtn), expireBeforeTimestampMs);
+  /** Expire snapshots on a given fully-qualified table name. */
+  public void expireSnapshots(String fqtn, long expireBeforeTimestampMs) {
+    expireSnapshots(getTable(fqtn), expireBeforeTimestampMs);
   }
 
-  /** Run ExpireSnapshots operation on a given {@link Table}. */
-  public ExpireSnapshots.Result expireSnapshots(Table table, long expireBeforeTimestampMs) {
-    return SparkActions.get(spark)
-        .expireSnapshots(table)
+  /** Expire snapshots on a given {@link Table}. */
+  public void expireSnapshots(Table table, long expireBeforeTimestampMs) {
+    table
+        .expireSnapshots()
+        .cleanExpiredFiles(false)
         .expireOlderThan(expireBeforeTimestampMs)
-        .deleteWith(
-            (file) -> {
-              // skip deletion
-              log.info("Detected file {} that is not part of survived snapshots", file);
-            })
-        .execute();
+        .commit();
   }
 
   /**

apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/SnapshotsExpirationSparkApp.java

@@ -1,16 +1,12 @@
 package com.linkedin.openhouse.jobs.spark;
 
 import com.linkedin.openhouse.jobs.spark.state.StateManager;
-import com.linkedin.openhouse.jobs.util.AppConstants;
-import io.opentelemetry.api.common.AttributeKey;
-import io.opentelemetry.api.common.Attributes;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.cli.CommandLine;
 import org.apache.commons.cli.Option;
-import org.apache.iceberg.actions.ExpireSnapshots;
 
 /**
  * Class with main entry point to run as a table snapshot expiration job. Snapshots for table which
@@ -40,42 +36,7 @@ protected void runInner(Operations ops) {
         granularity);
     long expireBeforeTimestampMs = System.currentTimeMillis() - TimeUnit.DAYS.toMillis(count);
     log.info("Expire snapshots before timestamp ms {}", expireBeforeTimestampMs);
-    ExpireSnapshots.Result result = ops.expireSnapshots(fqtn, expireBeforeTimestampMs);
-    log.info(
-        "Detected {} data files, {} manifest files, {} manifest list files that become orphaned as the result of snapshots expiration",
-        result.deletedDataFilesCount(),
-        result.deletedManifestsCount(),
-        result.deletedManifestListsCount());
-    METER
-        .counterBuilder(AppConstants.EXPIRED_FILE_COUNT)
-        .build()
-        .add(
-            result.deletedDataFilesCount(),
-            Attributes.of(
-                AttributeKey.stringKey(AppConstants.TABLE_NAME),
-                fqtn,
-                AttributeKey.stringKey(AppConstants.TYPE),
-                AppConstants.DATA_FILES));
-    METER
-        .counterBuilder(AppConstants.EXPIRED_FILE_COUNT)
-        .build()
-        .add(
-            result.deletedManifestsCount(),
-            Attributes.of(
-                AttributeKey.stringKey(AppConstants.TABLE_NAME),
-                fqtn,
-                AttributeKey.stringKey(AppConstants.TYPE),
-                AppConstants.MANIFEST_FILES));
-    METER
-        .counterBuilder(AppConstants.EXPIRED_FILE_COUNT)
-        .build()
-        .add(
-            result.deletedManifestListsCount(),
-            Attributes.of(
-                AttributeKey.stringKey(AppConstants.TABLE_NAME),
-                fqtn,
-                AttributeKey.stringKey(AppConstants.TYPE),
-                AppConstants.MANIFEST_LIST_FILES));
+    ops.expireSnapshots(fqtn, expireBeforeTimestampMs);
   }
 
   public static void main(String[] args) {

apps/spark/src/main/java/com/linkedin/openhouse/jobs/util/AppConstants.java

@@ -10,10 +10,6 @@ public final class AppConstants {
   // Spark App observability constants
   public static final String TYPE = "type";
   public static final String JOB_TYPE = "job_type";
-  public static final String DATA_FILES = "data_files";
-  public static final String MANIFEST_FILES = "manifest_files";
-  public static final String MANIFEST_LIST_FILES = "manifest_list_files";
-  public static final String EXPIRED_FILE_COUNT = "expired_file_count";
   public static final String ORPHAN_FILE_COUNT = "orphan_file_count";
   public static final String STAGED_FILE_COUNT = "staged_file_count";
   public static final String ORPHAN_DIRECTORY_COUNT = "orphan_directory_count";