Skip to content

Commit 3ef5eaf

Browse files
mkuchenbeckermkuchenbecker
committed
Merge mkuchenb/optimizer-3 + nest scheduler under apps/optimizer/scheduler
Brings the analyzer rename from opt-3 (apps/optimizer-analyzer → apps/optimizer/analyzer, package com.linkedin.openhouse.analyzer → com.linkedin.openhouse.optimizer.analyzer). Applies the same shape to the scheduler per PR #533 review: - apps/optimizer-scheduler/ → apps/optimizer/scheduler/ - package com.linkedin.openhouse.scheduler → com.linkedin.openhouse.optimizer.scheduler - Gradle module :apps:optimizer-scheduler → :apps:optimizer:scheduler settings.gradle conflict resolved to include both nested modules.
2 parents d7dead0 + eae3a4a commit 3ef5eaf

26 files changed

Lines changed: 218 additions & 159 deletions

File tree

‎apps/optimizer-analyzer/src/main/java/com/linkedin/openhouse/analyzer/CadenceBasedOrphanFilesDeletionAnalyzer.java‎

Lines changed: 0 additions & 51 deletions
This file was deleted.

apps/optimizer-analyzer/build.gradle renamed to apps/optimizer/analyzer/build.gradle

File renamed without changes.

apps/optimizer-analyzer/src/main/java/com/linkedin/openhouse/analyzer/AnalyzerApplication.java renamed to apps/optimizer/analyzer/src/main/java/com/linkedin/openhouse/optimizer/analyzer/AnalyzerApplication.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
package com.linkedin.openhouse.analyzer;
1+
package com.linkedin.openhouse.optimizer.analyzer;
22

33
import java.util.List;
44
import org.springframework.boot.CommandLineRunner;

apps/optimizer-analyzer/src/main/java/com/linkedin/openhouse/analyzer/AnalyzerRunner.java renamed to apps/optimizer/analyzer/src/main/java/com/linkedin/openhouse/optimizer/analyzer/AnalyzerRunner.java

Lines changed: 53 additions & 41 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
package com.linkedin.openhouse.analyzer;
1+
package com.linkedin.openhouse.optimizer.analyzer;
22

33
import com.linkedin.openhouse.optimizer.model.OperationTypeDto;
44
import com.linkedin.openhouse.optimizer.model.TableDto;
@@ -7,8 +7,6 @@
77
import com.linkedin.openhouse.optimizer.repository.TableOperationsHistoryRepository;
88
import com.linkedin.openhouse.optimizer.repository.TableOperationsRepository;
99
import com.linkedin.openhouse.optimizer.repository.TableStatsRepository;
10-
import java.time.Instant;
11-
import java.util.Comparator;
1210
import java.util.List;
1311
import java.util.Map;
1412
import java.util.Optional;
@@ -18,6 +16,7 @@
1816
import org.springframework.beans.factory.annotation.Value;
1917
import org.springframework.data.domain.PageRequest;
2018
import org.springframework.stereotype.Component;
19+
import org.springframework.transaction.annotation.Transactional;
2120

2221
/**
2322
* Core analysis loop. For one operation type per call, iterates databases and evaluates each table
@@ -28,9 +27,8 @@
2827
* tables); past that the per-db query shape and projection need further tuning. Scale-up work is
2928
* tracked in <a href="https://linkedin.atlassian.net/browse/BDP-102182">BDP-102182</a>.
3029
*
31-
* <p>// TODO(scale-test): benchmark the per-db working set at up to 10k tables and measure JVM heap
32-
* residency for the three intermediate maps; per-db iteration bounds memory by tables-per-db rather
33-
* than tables-total, but the upper bound still needs empirical validation.
30+
* <p>The per-db working-set upper bound is not yet empirically validated. Scale-test tracked in <a
31+
* href="https://linkedin.atlassian.net/browse/BDP-102738">BDP-102738</a>.
3432
*/
3533
@Slf4j
3634
@Component
@@ -42,6 +40,8 @@ public class AnalyzerRunner {
4240
private final TableOperationsRepository operationsRepo;
4341
private final TableOperationsHistoryRepository historyRepo;
4442

43+
// Inline default also set on the field so Mockito-constructed instances (no Spring context) get
44+
// a usable value; with Spring, the @Value annotation overrides this.
4545
@Value("${optimizer.repo.default-limit:10000}")
4646
private int defaultLimit = 10_000;
4747

@@ -78,21 +78,19 @@ public void analyze(
7878
log.info("Analysis complete for {}", operationType);
7979
}
8080

81-
private void analyzeDatabase(
81+
@Transactional
82+
void analyzeDatabase(
8283
OperationAnalyzer analyzer,
8384
String databaseName,
8485
Optional<String> tableName,
8586
Optional<String> tableUuid) {
8687

87-
com.linkedin.openhouse.optimizer.db.OperationType dbOperationType =
88-
analyzer.getOperationType().toDb();
89-
9088
// Pre-load the small sides of the joins — bounded by tables in this database.
9189
PageRequest page = PageRequest.of(0, defaultLimit);
9290
Map<String, TableOperationDto> currentOps =
9391
operationsRepo
9492
.find(
95-
Optional.of(dbOperationType),
93+
Optional.of(analyzer.getOperationType().toDb()),
9694
Optional.empty(),
9795
tableUuid,
9896
Optional.of(databaseName),
@@ -108,14 +106,14 @@ private void analyzeDatabase(
108106
TableOperationDto::getTableUuid, op -> op, TableOperationDto::mostRecent));
109107

110108
Map<String, TableOperationsHistoryDto> latestHistory =
111-
historyRepo.findLatest(dbOperationType, page).stream()
109+
historyRepo.findLatest(analyzer.getOperationType().toDb(), page).stream()
112110
.filter(r -> r.getTableUuid() != null)
113111
.map(TableOperationsHistoryDto::fromRow)
114112
.collect(
115113
Collectors.toMap(
116114
TableOperationsHistoryDto::getTableUuid,
117115
h -> h,
118-
AnalyzerRunner::moreRecentHistory));
116+
TableOperationsHistoryDto::after));
119117

120118
List<TableDto> tables =
121119
statsRepo.find(Optional.of(databaseName), tableName, tableUuid, page).stream()
@@ -126,39 +124,53 @@ private void analyzeDatabase(
126124
/*
127125
* For each table in this database, decide whether to create a new PENDING operation.
128126
*
129-
* 1. Skip tables not opted in to this operation type. The opt-in check today reads a
130-
* table-property flag; in the future it will read a denormalized column.
127+
* 1. Skip tables not opted in to this operation type.
131128
* 2. Look up the table's current active operation (if any) and its most recent completed
132129
* history entry from the maps loaded above.
133130
* 3. Delegate the schedule-or-not decision to the analyzer's shouldSchedule — strategy
134131
* encapsulates cadence, retry policy, and any future per-operation signals.
135132
* 4. On true, persist a new PENDING operation. The scheduler picks it up on its next pass.
136133
*/
137-
tables.forEach(
138-
table -> {
139-
if (!analyzer.isEnabled(table)) {
140-
return;
141-
}
142-
Optional<TableOperationDto> currentOp =
143-
Optional.ofNullable(currentOps.get(table.getTableUuid()));
144-
Optional<TableOperationsHistoryDto> entry =
145-
Optional.ofNullable(latestHistory.get(table.getTableUuid()));
146-
if (analyzer.shouldSchedule(table, currentOp, entry)) {
147-
TableOperationDto op = TableOperationDto.pending(table, analyzer.getOperationType());
148-
operationsRepo.save(op.toRow());
149-
log.info(
150-
"Created PENDING {} operation for table {}.{}",
151-
analyzer.getOperationType(),
152-
table.getDatabaseName(),
153-
table.getTableId());
154-
}
155-
});
156-
}
157-
158-
private static TableOperationsHistoryDto moreRecentHistory(
159-
TableOperationsHistoryDto a, TableOperationsHistoryDto b) {
160-
Comparator<TableOperationsHistoryDto> byCompletedAt =
161-
Comparator.comparing(r -> r.getCompletedAt() != null ? r.getCompletedAt() : Instant.EPOCH);
162-
return byCompletedAt.compare(a, b) >= 0 ? a : b;
134+
int created = 0;
135+
int failed = 0;
136+
for (TableDto table : tables) {
137+
if (!analyzer.isEnabled(table)) {
138+
continue;
139+
}
140+
Optional<TableOperationDto> currentOp =
141+
Optional.ofNullable(currentOps.get(table.getTableUuid()));
142+
Optional<TableOperationsHistoryDto> entry =
143+
Optional.ofNullable(latestHistory.get(table.getTableUuid()));
144+
if (!analyzer.shouldSchedule(table, currentOp, entry)) {
145+
continue;
146+
}
147+
try {
148+
TableOperationDto op = TableOperationDto.pending(table, analyzer.getOperationType());
149+
operationsRepo.save(op.toRow());
150+
log.debug(
151+
"Created PENDING {} operation for table {}.{}",
152+
analyzer.getOperationType(),
153+
table.getDatabaseName(),
154+
table.getTableId());
155+
created++;
156+
} catch (RuntimeException e) {
157+
// One bad table should not abort the rest of the database. Log and continue; the next
158+
// analyzer pass will retry for any table whose save failed here.
159+
log.error(
160+
"Failed to create PENDING {} operation for table {}.{}: {}",
161+
analyzer.getOperationType(),
162+
table.getDatabaseName(),
163+
table.getTableId(),
164+
e.toString(),
165+
e);
166+
failed++;
167+
}
168+
}
169+
log.info(
170+
"Database {}: created {} PENDING {} operation(s) ({} failed)",
171+
databaseName,
172+
created,
173+
analyzer.getOperationType(),
174+
failed);
163175
}
164176
}

‎apps/optimizer/analyzer/src/main/java/com/linkedin/openhouse/optimizer/analyzer/CadenceBasedOrphanFilesDeletionAnalyzer.java‎

Lines changed: 83 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,83 @@
1+
package com.linkedin.openhouse.optimizer.analyzer;
2+
3+
import com.linkedin.openhouse.optimizer.model.OperationTypeDto;
4+
import com.linkedin.openhouse.optimizer.model.TableDto;
5+
import com.linkedin.openhouse.optimizer.model.TableOperationDto;
6+
import com.linkedin.openhouse.optimizer.model.TableOperationsHistoryDto;
7+
import java.time.Duration;
8+
import java.util.Optional;
9+
import org.springframework.beans.factory.annotation.Value;
10+
import org.springframework.stereotype.Component;
11+
12+
/**
13+
* Decides when to schedule an Orphan-Files-Deletion (OFD) run for a table.
14+
*
15+
* <p>OFD removes data files in the table's storage directory that are no longer referenced by any
16+
* Iceberg snapshot — left-over output from failed writes, expired snapshots, or interrupted
17+
* compactions. Running it too often wastes compute; running it too rarely lets orphan files
18+
* accumulate and bloats storage cost. This analyzer balances the two on a per-table cadence.
19+
*
20+
* <h2>When OFD fires for a table</h2>
21+
*
22+
* All of the following must be true:
23+
*
24+
* <ol>
25+
* <li><b>Opt-in.</b> The table sets {@code maintenance.optimizer.ofd.enabled=true} in its table
26+
* properties. Without this flag, the analyzer ignores the table entirely.
27+
* <li><b>No active operation already in flight.</b> If the table has a non-CANCELED operation row
28+
* (PENDING, SCHEDULING, or SCHEDULED), the scheduler already owns it and the analyzer stays
29+
* out. A CANCELED row does not block — it is treated as if no operation exists.
30+
* <li><b>Cadence elapsed since the last completed run.</b>
31+
* <ul>
32+
* <li>If the table has <i>no</i> prior history, schedule immediately.
33+
* <li>If the most recent history entry is {@code SUCCESS}, wait {@code
34+
* ofd.success-retry-hours} (default 16h) after its {@code completedAt} before
35+
* scheduling again. Set below 24h so that even when a run lands at an unlucky time of
36+
* day, at least one re-evaluation is guaranteed within any rolling 24-hour window.
37+
* <li>If the most recent history entry is {@code FAILED}, wait {@code
38+
* ofd.failure-retry-hours} (default 1h) before retrying — shorter than the success
39+
* interval so transient failures recover quickly.
40+
* </ul>
41+
* </ol>
42+
*
43+
* <p>The two retry intervals are configurable via {@code application.properties} and can be tuned
44+
* per environment. The opt-in property is per-table and managed through the standard table-
45+
* properties API.
46+
*/
47+
@Component
48+
public class CadenceBasedOrphanFilesDeletionAnalyzer implements OperationAnalyzer {
49+
50+
static final String OFD_ENABLED_PROPERTY = "maintenance.optimizer.ofd.enabled";
51+
52+
private final CadencePolicy cadencePolicy;
53+
54+
public CadenceBasedOrphanFilesDeletionAnalyzer(
55+
@Value("${ofd.success-retry-hours:16}") long successRetryHours,
56+
@Value("${ofd.failure-retry-hours:1}") long failureRetryHours) {
57+
this.cadencePolicy =
58+
new CadencePolicy(Duration.ofHours(successRetryHours), Duration.ofHours(failureRetryHours));
59+
}
60+
61+
/** Package-private for tests that supply a pre-built {@link CadencePolicy}. */
62+
CadenceBasedOrphanFilesDeletionAnalyzer(CadencePolicy cadencePolicy) {
63+
this.cadencePolicy = cadencePolicy;
64+
}
65+
66+
@Override
67+
public OperationTypeDto getOperationType() {
68+
return OperationTypeDto.ORPHAN_FILES_DELETION;
69+
}
70+
71+
@Override
72+
public boolean isEnabled(TableDto table) {
73+
return "true".equals(table.getTableProperties().get(OFD_ENABLED_PROPERTY));
74+
}
75+
76+
@Override
77+
public boolean shouldSchedule(
78+
TableDto table,
79+
Optional<TableOperationDto> currentOp,
80+
Optional<TableOperationsHistoryDto> latestHistory) {
81+
return cadencePolicy.shouldSchedule(currentOp, latestHistory);
82+
}
83+
}

apps/optimizer-analyzer/src/main/java/com/linkedin/openhouse/analyzer/CadencePolicy.java renamed to apps/optimizer/analyzer/src/main/java/com/linkedin/openhouse/optimizer/analyzer/CadencePolicy.java

Lines changed: 21 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
package com.linkedin.openhouse.analyzer;
1+
package com.linkedin.openhouse.optimizer.analyzer;
22

33
import com.linkedin.openhouse.optimizer.model.HistoryStatusDto;
44
import com.linkedin.openhouse.optimizer.model.OperationStatusDto;
@@ -23,8 +23,9 @@ public class CadencePolicy {
2323

2424
/**
2525
* How long to wait after a successful operation before re-evaluating the table. For example, if
26-
* set to 24 hours and OFD succeeded at 10:00 AM Monday, the table won't be scheduled again until
27-
* after 10:00 AM Tuesday.
26+
* set to 16 hours and OFD succeeded at 10:00 AM Monday, the table becomes eligible again at 2:00
27+
* AM Tuesday. Configured below 24h so that at least one re-evaluation is guaranteed within any
28+
* rolling 24-hour window regardless of when the prior run landed.
2829
*/
2930
private final Duration successRetryInterval;
3031

@@ -50,8 +51,22 @@ public boolean shouldSchedule(
5051
}
5152

5253
private boolean readyAfterHistoryEntry(TableOperationsHistoryDto entry) {
53-
Duration interval =
54-
entry.getStatus() == HistoryStatusDto.FAILED ? failureRetryInterval : successRetryInterval;
55-
return Duration.between(entry.getCompletedAt(), Instant.now()).compareTo(interval) > 0;
54+
return Duration.between(entry.getCompletedAt(), Instant.now())
55+
.compareTo(intervalFor(entry.getStatus()))
56+
> 0;
57+
}
58+
59+
private Duration intervalFor(HistoryStatusDto status) {
60+
// Explicit per-status mapping. Adding a new HistoryStatusDto value forces this switch to
61+
// grow a case; the default throws so an un-handled value surfaces at runtime rather than
62+
// silently falling into the success bucket.
63+
switch (status) {
64+
case SUCCESS:
65+
return successRetryInterval;
66+
case FAILED:
67+
return failureRetryInterval;
68+
default:
69+
throw new IllegalStateException("Unhandled HistoryStatusDto value: " + status);
70+
}
5671
}
5772
}

apps/optimizer-analyzer/src/main/java/com/linkedin/openhouse/analyzer/OperationAnalyzer.java renamed to apps/optimizer/analyzer/src/main/java/com/linkedin/openhouse/optimizer/analyzer/OperationAnalyzer.java

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
package com.linkedin.openhouse.analyzer;
1+
package com.linkedin.openhouse.optimizer.analyzer;
22

33
import com.linkedin.openhouse.optimizer.model.OperationTypeDto;
44
import com.linkedin.openhouse.optimizer.model.TableDto;
@@ -10,7 +10,7 @@
1010
* Strategy interface for a single operation type. Each implementation decides whether a given table
1111
* needs an operation recommendation upserted in the Optimizer Service.
1212
*
13-
* <p>// TODO(circuit-breaker): a chronically-failing table currently produces a new PENDING row on
13+
* <p>TODO(circuit-breaker): a chronically-failing table currently produces a new PENDING row on
1414
* every Analyzer pass. Add a circuit breaker that suppresses scheduling for a (table, type) after N
1515
* consecutive FAILED history entries. Requirements: configurable threshold per operation type,
1616
* automatic reset via exponential backoff so tables can recover, and an operator-visible signal

apps/optimizer-analyzer/src/main/resources/application.properties renamed to apps/optimizer/analyzer/src/main/resources/application.properties

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,6 @@ spring.datasource.url=${OPTIMIZER_DB_URL:jdbc:h2:mem:analyzerdb;DB_CLOSE_DELAY=-
44
spring.datasource.username=${OPTIMIZER_DB_USER:sa}
55
spring.datasource.password=${OPTIMIZER_DB_PASSWORD:}
66
spring.jpa.hibernate.ddl-auto=none
7-
ofd.success-retry-hours=24
7+
ofd.success-retry-hours=16
88
ofd.failure-retry-hours=1
99
optimizer.repo.default-limit=10000

apps/optimizer-analyzer/src/test/java/com/linkedin/openhouse/analyzer/AnalyzerRunnerTest.java renamed to apps/optimizer/analyzer/src/test/java/com/linkedin/openhouse/optimizer/analyzer/AnalyzerRunnerTest.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
package com.linkedin.openhouse.analyzer;
1+
package com.linkedin.openhouse.optimizer.analyzer;
22

33
import static org.assertj.core.api.Assertions.assertThat;
44
import static org.mockito.ArgumentMatchers.any;

0 commit comments

Comments
 (0)