Optimizer: Batched orphan file deletion using bin packing by abhisheknath2011 · Pull Request #599 · linkedin/openhouse

abhisheknath2011 · 2026-05-22T23:02:03Z

Summary

Introduces BatchedOrphanFilesDeletionSparkApp, the multi-table counterpart of the existing single-table OrphanFilesDeletionSparkApp. One Spark job now processes a list of (table, operationId) pairs that the optimizer scheduler bin-packed into a single batch, reporting SUCCESS/FAILED per operation back to the Optimizer Service.

Also lands a first-fit-decreasing bin packer (jobs.util.binpack) that the scheduler (#534) will use to assemble those batches. The packer has no caller in this PR — it ships alongside the Spark app so the algorithm can be reviewed independently from the scheduler wiring. Keeping it in apps/spark for now since the scheduler module isn't merged yet; it can move along side the scheduler once PR #534 is merged.

Key design choices:

Per-table failure isolation — exceptions in one table are caught, FAILED is posted for that operationId, and remaining tables continue. The job exits 0 if at least one table succeeds.
Recoverable result reporting — if POST /v1/optimizer/operations/update itself fails, the row stays SCHEDULED and the Analyzer's stale-timeout will re-queue it. No retry storms in the Spark driver.
Scheduler decides parallelism, not the app — --driverParallelism is honoured verbatim; the app does not pick its own thread count.
Bin packer never drops oversized tables — an item exceeding any single cap is placed in a dedicated bin rather than silently skipped.
Self-contained wire DTO — OperationUpdateRequest is mirrored in apps/spark so this PR compiles independently of the optimizer-service module's merge order. Replace with the shared DTO once the optimizer module lives in apps/.
Bin packer (com.linkedin.openhouse.jobs.util.binpack):
-BinItem — fqtn, operationId, tableUuid, db/table, weight (numFiles), sizeBytes
Bin — mutable accumulator with three-dimensional fit check
FirstFitDecreasingBinPacker — FFD by weight with secondary caps on bytes and item count; oversized items get a dedicated bin

Optimizer Service client (com.linkedin.openhouse.jobs.spark.optimizer):

OperationUpdateRequest — wire-compatible body for POST /v1/optimizer/operations/update
OptimizerServiceClient — thin OkHttp client with sensible timeouts

Batched Spark app (com.linkedin.openhouse.jobs.spark):

BatchedOrphanFilesDeletionSparkApp — extends BaseSparkApp; iterates entries via a fixed thread pool, reuses Operations.deleteOrphanFiles(...) per table, posts per-operation status, runs the existing TableStateValidator per table

CLI:
--tableNames db.t1,db.t2,db.t3
--operationIds op-uuid-1,op-uuid-2,op-uuid-3
--tableUuids tab-uuid-1,tab-uuid-2,tab-uuid-3
--resultsEndpoint http://optimizer.svc:8080
--driverParallelism 4
plus existing OFD knobs (--ttl, --backupDir, --concurrentDeletes, --streamResults, --maxOrphanFileSampleSize).

Issue] Briefly discuss the summary of the changes made in this
pull request in 2-3 lines.

Changes

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

Testing Done

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.
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.

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.

Open items for reviewers:
- OperationUpdateRequest is a local mirror of feat(optimizer): [2/N] Optimizer REST Service and Controller #531's UpdateOperationRequest. Should we introduce an apps/optimizer shared module here (mirroring the analyzer's pattern in feat(optimizer): [3/N] Analyzer #533) and depend on it instead?
- Bin packer placement: keep in apps/spark/util/binpack or move to the scheduler module in feat(optimizer): [4/N] Scheduler app #534 alongside its only caller?
- Wire shape uses POST /v1/optimizer/operations/update per feat(optimizer): [2/N] Optimizer REST Service and Controller #531. If that endpoint changes name (e.g. to POST /{id}/complete), OptimizerServiceClient.UPDATE_PATH is the only place that needs to change.

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

mkuchenbecker · 2026-05-22T23:11:51Z

+        .connectTimeout(10, TimeUnit.SECONDS)
+        .readTimeout(30, TimeUnit.SECONDS)
+        .writeTimeout(30, TimeUnit.SECONDS)


Ideally these would be configs.

mkuchenbecker · 2026-05-22T23:13:24Z

+  private static String stripTrailingSlash(String url) {
+    if (url == null || url.isEmpty()) {
+      throw new IllegalArgumentException("Optimizer Service base URL must be non-empty");
+    }
+    return url.endsWith("/") ? url.substring(0, url.length() - 1) : url;
+  }


(1) Seems heavyweight
(2) Can we just assume not null or attempt to use optional instead if null might be present? Under what conditions is the url valid to be null?

mkuchenbecker · 2026-05-22T23:15:33Z

+  private final List<BatchEntry> entries;
+  private final String resultsEndpoint;
+  private final int driverParallelism;
+  private final long ttlSeconds;
+  private final String backupDir;
+  private final int concurrentDeletes;
+  private final boolean streamResults;
+  private final int maxOrphanFileSampleSize;


Thoughts on a OFD parameters object that we use a builder to construct? The main thought was to encapsulate parameters rather than having them be manually maintained. I'm generally a fan of generating with annotations to avoid this boilerplate as lines 62-92 are all just defining parameters and a public funciton to supply them.

mkuchenbecker · 2026-05-22T23:17:44Z

+      int concurrentDeletes,
+      boolean streamResults,
+      int maxOrphanFileSampleSize) {
+    super(jobId, stateManager, otelEmitter);


There is a callback to complete job on HTS right? Do we need to adapt that or is it fine to leave as-is?

Inherited from BaseSparkApp.run(): stateManager.updateState(jobId, SUCCEEDED/FAILED) already fires via onStarted/onFinished plus heartbeats — HTS sees this job's lifecycle. The new optimizer-side callbacks are per-operation (per-table); HTS callback is per-job. Both are intentional and orthogonal.

mkuchenbecker · 2026-05-22T23:26:46Z

+      try {
+        client.updateOperation(body);
+      } catch (IOException e) {
+        log.error(


counter? We can get signal on how many jobs are failing to emit metrics.

Yes emitting counter metrics and logging error.

mkuchenbecker · 2026-05-22T23:28:20Z

+    private int countOrphans(DeleteOrphanFiles.Result result) {
+      int count = 0;
+      for (String unused : result.orphanFileLocations()) {
+        count++;
+      }
+      return count;
+    }
+  }


Can we do result.count()?

Use iterables count to reduce driver memory usage as we don't need full path materialization.

mkuchenbecker · 2026-05-22T23:31:24Z

+    if (tableNames == null
+        || operationIds == null
+        || tableUuids == null
+        || tableNames.isEmpty()


Is there a practical limit to the number of tables in the job based on the input string length limits?

## Optimizer Stack | PR | Content | |---|---| | #527 | Data Model | | #530 | Database Repos | | #531 | REST service | | #533 **(this)** | Analyzer app | | #534 | Scheduler app | | #599 | Spark BatchedOFD app | | #tbd | Infra, docker-compose, smoke test | ## Summary PR 3 of N in the optimizer stack. Introduces `apps/optimizer-analyzer`, a Spring Boot CommandLineRunner that evaluates every table in `table_stats` against pluggable `OperationAnalyzer` strategies. The first strategy, `OrphanFilesDeletionAnalyzer`, schedules OFD operations with 24h success / 1h failure retry cadence, a 6h SCHEDULED timeout, and a 5-strike circuit breaker. Key design choices: - Bulk-loads operations and history into maps (one query per type), then iterates the stats list — O(types) queries, not O(tables). - Uses the existing generic `find()` repository methods with null params. - Pure unit tests with Mockito — no Spring context needed. ## Changes - [ ] Client-facing API Changes - [ ] Internal API Changes - [ ] Bug Fixes - [x] New Features - [ ] Performance Improvements - [ ] Code Style - [ ] Refactoring - [ ] Documentation - [x] Tests **Core**: `AnalyzerRunner` — loads table_stats, pre-loads operations and history into maps, evaluates each table against all analyzers, circuit breaker logic. **Strategy interface**: `OperationAnalyzer` — `isEnabled(table)`, `shouldSchedule(table, currentOp, latestHistory)`, `getCircuitBreakerThreshold()`. **Cadence policy**: `CadencePolicy` — encapsulates time-based retry logic shared across operation types. **OFD analyzer**: `OrphanFilesDeletionAnalyzer` — enabled via `maintenance.optimizer.ofd.enabled` table property. ## Testing Done - [ ] Manually Tested on local docker setup. Please include commands ran, and their output. - [x] Added new tests for the changes made. - [ ] Updated existing tests to reflect the changes made. - [ ] 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. 25 unit tests: - `AnalyzerRunnerTest` (7 tests) — eligible table insertion, cadence skip, disabled table, shouldSchedule=false, null UUID, circuit breaker trip, below-threshold pass - `OrphanFilesDeletionAnalyzerTest` (18 tests) — isEnabled variants, shouldSchedule for no-op/PENDING/SCHEDULING/SCHEDULED with history combinations ``` ./gradlew :apps:optimizer-analyzer:test # BUILD SUCCESSFUL — 25 tests pass ``` # Additional Information - [ ] Breaking Changes - [ ] Deprecations - [x] Large PR broken into smaller PRs, and PR plan linked in the description. --------- Co-authored-by: mkuchenbecker <mkuchenbecker@users.noreply.github.com> Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com> Co-authored-by: Abhishek Nath <anath1@linkedin.com>

Optimizer: Batched orphan file deletion using bin packing

200508e

mkuchenbecker reviewed May 22, 2026

View reviewed changes

mkuchenbecker mentioned this pull request May 22, 2026

feat(optimizer): [3/N] Analyzer #533

Merged

17 tasks

abhisheknath2011 added 2 commits May 26, 2026 15:20

Addressed review comments

ff6c881

Count orphan files using Iterables to reduce driver memeory usage

ec76920

abhisheknath2011 mentioned this pull request May 27, 2026

Integrate batched orphan files deletion with the existing schedule workflow #604

Draft

17 tasks

abhisheknath2011 marked this pull request as ready for review May 27, 2026 19:58

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Optimizer: Batched orphan file deletion using bin packing#599

Optimizer: Batched orphan file deletion using bin packing#599
abhisheknath2011 wants to merge 3 commits into
linkedin:mainfrom
abhisheknath2011:batched-ofd

abhisheknath2011 commented May 22, 2026 •

edited

Loading

Uh oh!

mkuchenbecker May 22, 2026

Uh oh!

mkuchenbecker May 22, 2026

Uh oh!

mkuchenbecker May 22, 2026

Uh oh!

mkuchenbecker May 22, 2026

Uh oh!

abhisheknath2011 May 27, 2026

Uh oh!

Uh oh!

Uh oh!

mkuchenbecker May 22, 2026

Uh oh!

abhisheknath2011 May 27, 2026

Uh oh!

Uh oh!

mkuchenbecker May 22, 2026

Uh oh!

abhisheknath2011 May 27, 2026

Uh oh!

mkuchenbecker May 22, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Conversation

abhisheknath2011 commented May 22, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Changes

Testing Done

Additional Information

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

abhisheknath2011 commented May 22, 2026 •

edited

Loading