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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged

mkuchenbecker merged 209 commits into main from mkuchenb/optimizer-3

May 27, 2026

apps/optimizer/analyzerapp/build.gradle

-Original file line number
+Diff line change
@@ -0,0 +1,14 @@
+    plugins {
+      id 'openhouse.springboot-ext-conventions'
+      id 'org.springframework.boot' version '2.7.8'
+    }
+    // Deployable Spring Boot wrapper around the analyzer library. Holds AnalyzerApplication (the
+    // @SpringBootApplication entry point) and application.properties; the analysis logic lives in
+    // :services:optimizer:analyzer.
+    dependencies {
+      implementation project(':services:optimizer:analyzer')
+      implementation 'org.springframework.boot:spring-boot-starter:2.7.8'
+      implementation 'org.springframework.boot:spring-boot-starter-data-jpa:2.7.8'
+      runtimeOnly 'mysql:mysql-connector-java:8.0.33'
+    }

...lyzerapp/src/main/java/com/linkedin/openhouse/optimizer/analyzer/AnalyzerApplication.java

-Original file line number
+Diff line change
@@ -0,0 +1,29 @@
+    package com.linkedin.openhouse.optimizer.analyzer;
+    import java.util.List;
+    import org.springframework.boot.CommandLineRunner;
+    import org.springframework.boot.SpringApplication;
+    import org.springframework.boot.autoconfigure.SpringBootApplication;
+    import org.springframework.boot.autoconfigure.domain.EntityScan;
+    import org.springframework.context.annotation.Bean;
+    import org.springframework.data.jpa.repository.config.EnableJpaRepositories;
+    /** Entry point for the Optimizer Analyzer application. */
+    @SpringBootApplication
+    @EntityScan(basePackages = "com.linkedin.openhouse.optimizer.db")
+    @EnableJpaRepositories(basePackages = "com.linkedin.openhouse.optimizer.repository")
+    public class AnalyzerApplication {
+      public static void main(String[] args) {
+        SpringApplication.run(AnalyzerApplication.class, args);
+      }
+      /**
+       * Runs the analyzer once per registered {@link OperationAnalyzer} per process invocation. Each
+       * call is scoped to one operation type; the runner iterates databases internally.
+       */
+      @Bean
+      public CommandLineRunner run(AnalyzerRunner runner, List<OperationAnalyzer> analyzers) {
+        return args -> analyzers.forEach(a -> runner.analyze(a.getOperationType()));
+      }
+    }

apps/optimizer/analyzerapp/src/main/resources/application.properties

-Original file line number
+Diff line change
@@ -0,0 +1,8 @@
+    spring.application.name=openhouse-optimizer-analyzer
+    spring.main.web-application-type=none
+    spring.datasource.url=${OPTIMIZER_DB_URL:jdbc:h2:mem:analyzerdb;DB_CLOSE_DELAY=-1;MODE=MySQL}
+    spring.datasource.username=${OPTIMIZER_DB_USER:sa}
+    spring.datasource.password=${OPTIMIZER_DB_PASSWORD:}
+    spring.jpa.hibernate.ddl-auto=none
+    ofd.success-retry-hours=16
+    ofd.failure-retry-hours=1

services/optimizer/analyzer/build.gradle

-Original file line number
+Diff line change
@@ -0,0 +1,34 @@
+    plugins {
+      id 'openhouse.springboot-ext-conventions'
+      id 'org.springframework.boot' version '2.7.8'
+    }
+    // Library jar — the @SpringBootApplication entry point lives in :apps:optimizer:analyzerapp.
+    // Disable bootJar so we don't try to assemble a runnable jar from a library that has no main
+    // class; keep jar enabled so consumers (the apps wrapper) get a normal library artifact.
+    bootJar {
+      enabled = false
+    }
+    jar {
+      enabled = true
+      archiveClassifier = ''
+    }
+    dependencies {
+      // api: the analyzer's public types (e.g. OperationAnalyzer's signature, OperationTypeDto) come
+      // from :services:optimizer, so consumers of this library see them on their compile classpath.
+      api project(':services:optimizer')
+      implementation 'org.springframework.boot:spring-boot-starter:2.7.8'
+      implementation 'org.springframework.boot:spring-boot-starter-webflux:2.7.8'
+      implementation 'org.springframework.boot:spring-boot-starter-data-jpa:2.7.8'
+      implementation 'org.springframework.boot:spring-boot-starter-aop:2.7.8'
+      runtimeOnly 'mysql:mysql-connector-java:8.0.33'
+      testImplementation 'org.springframework.boot:spring-boot-starter-test:2.7.8'
+      testImplementation 'com.squareup.okhttp3:mockwebserver:4.10.0'
+      testRuntimeOnly 'com.h2database:h2'
+    }
+    test {
+      useJUnitPlatform()
+    }

...izer/analyzer/src/main/java/com/linkedin/openhouse/optimizer/analyzer/AnalyzerRunner.java

-Original file line number
+Diff line change
@@ -0,0 +1,172 @@
+    package com.linkedin.openhouse.optimizer.analyzer;
+    import com.linkedin.openhouse.optimizer.model.OperationTypeDto;
+    import com.linkedin.openhouse.optimizer.model.TableDto;
+    import com.linkedin.openhouse.optimizer.model.TableOperationDto;
+    import com.linkedin.openhouse.optimizer.model.TableOperationsHistoryDto;
+    import com.linkedin.openhouse.optimizer.repository.TableOperationsHistoryRepository;
+    import com.linkedin.openhouse.optimizer.repository.TableOperationsRepository;
+    import com.linkedin.openhouse.optimizer.repository.TableStatsRepository;
+    import java.util.List;
+    import java.util.Map;
+    import java.util.Optional;
+    import java.util.stream.Collectors;
+    import lombok.RequiredArgsConstructor;
+    import lombok.extern.slf4j.Slf4j;
+    import org.springframework.data.domain.Pageable;
+    import org.springframework.stereotype.Component;
+    import org.springframework.transaction.annotation.Transactional;
+    /**
+     * Core analysis loop. For one operation type per call, iterates databases and evaluates each table
+     * in a database against the matching {@link OperationAnalyzer}.
+     *
+     * <p>Both sides of the join — current operations and latest history per (table, type) — are loaded
+     * into maps once per database before the table loop. This is correct at small scale (≤~100k
+     * tables); past that the per-db query shape and projection need further tuning.
+     *
+     * <p>The per-db working-set upper bound is not yet empirically validated.
+     */
+    @Slf4j
+    @Component
+    @RequiredArgsConstructor
+    public class AnalyzerRunner {
+      private final List<OperationAnalyzer> analyzers;
+      private final TableStatsRepository statsRepo;
+      private final TableOperationsRepository operationsRepo;
+      private final TableOperationsHistoryRepository historyRepo;
+      /**
+       * Run the analysis loop for {@code operationType} across all databases, with no filters.
+       * Equivalent to {@link #analyze(OperationTypeDto, Optional, Optional, Optional)} with all-empty
+       * filters.
+       */
+      public void analyze(OperationTypeDto operationType) {
+        analyze(operationType, Optional.empty(), Optional.empty(), Optional.empty());
+      }
+      /**
+       * Run the analysis loop for the given operation type, optionally scoped to a single database,
+       * table name, or table UUID. Iterates databases one at a time so the working set is bounded by
+       * tables-per-db, not tables-total.
+       */
+      public void analyze(
+          OperationTypeDto operationType,
+          Optional<String> databaseName,
+          Optional<String> tableName,
+          Optional<String> tableUuid) {
+        OperationAnalyzer analyzer =
+            analyzers.stream()
+                .filter(a -> a.getOperationType() == operationType)
+                .findFirst()
+                .orElseThrow(
+                    () ->
+                        new IllegalStateException(
+                            "No analyzer registered for operation type " + operationType));
+        List<String> dbs = databaseName.map(List::of).orElseGet(statsRepo::findDistinctDatabaseNames);
+        log.info("Analyzing {} across {} database(s)", operationType, dbs.size());
+        dbs.forEach(db -> analyzeDatabase(analyzer, db, tableName, tableUuid));
+        log.info("Analysis complete for {}", operationType);
+      }
+      @Transactional
+      void analyzeDatabase(
+          OperationAnalyzer analyzer,
+          String databaseName,
+          Optional<String> tableName,
+          Optional<String> tableUuid) {
+        // Load the three join inputs unbounded for this database. Aligned page-by-page pagination on
+        // these maps would leave keys in one map's page mismatched with the others' — a table whose
+        // op/history happens to fall in a different page would be misread as "no current op / no
+        // history" and trigger duplicate scheduling. Correctness requires the maps to be complete
+        // relative to the tables being processed; the working set is bounded by tables-in-db, not by
+        // any per-cycle cap.
+        Map<String, TableOperationDto> currentOps =
+            operationsRepo
+                .find(
+                    Optional.of(analyzer.getOperationType().toDb()),
+                    Optional.empty(),
+                    tableUuid,
+                    Optional.of(databaseName),
+                    tableName,
+                    Optional.empty(),
+                    Optional.empty(),
+                    Pageable.unpaged())
+                .stream()
+                .filter(e -> e.getTableUuid() != null)
+                .map(TableOperationDto::fromRow)
+                .collect(
+                    Collectors.toMap(
+                        TableOperationDto::getTableUuid, op -> op, TableOperationDto::mostRecent));
+        Map<String, TableOperationsHistoryDto> latestHistory =
+            historyRepo.findLatest(analyzer.getOperationType().toDb(), Pageable.unpaged()).stream()
+                .filter(r -> r.getTableUuid() != null)
+                .map(TableOperationsHistoryDto::fromRow)
+                .collect(
+                    Collectors.toMap(
+                        TableOperationsHistoryDto::getTableUuid,
+                        h -> h,
+                        TableOperationsHistoryDto::after));
+        List<TableDto> tables =
+            statsRepo.find(Optional.of(databaseName), tableName, tableUuid, Pageable.unpaged()).stream()
+                .filter(row -> row.getTableUuid() != null)
+                .map(TableDto::fromRow)
+                .collect(Collectors.toList());
+        /*
+         * For each table in this database, decide whether to create a new PENDING operation.
+         *
+         * 1. Skip tables not opted in to this operation type.
+         * 2. Look up the table's current active operation (if any) and its most recent completed
+         *    history entry from the maps loaded above.
+         * 3. Delegate the schedule-or-not decision to the analyzer's shouldSchedule — strategy
+         *    encapsulates cadence, retry policy, and any future per-operation signals.
+         * 4. On true, persist a new PENDING operation. The scheduler picks it up on its next pass.
+         */
+        int created = 0;
+        int failed = 0;
+        for (TableDto table : tables) {
+          if (!analyzer.isEnabled(table)) {
+            continue;
+          }
+          Optional<TableOperationDto> currentOp =
+              Optional.ofNullable(currentOps.get(table.getTableUuid()));
+          Optional<TableOperationsHistoryDto> entry =
+              Optional.ofNullable(latestHistory.get(table.getTableUuid()));
+          if (!analyzer.shouldSchedule(table, currentOp, entry)) {
+            continue;
+          }
+          try {
+            TableOperationDto op = TableOperationDto.pending(table, analyzer.getOperationType());
+            operationsRepo.save(op.toRow());
+            log.debug(
+                "Created PENDING {} operation for table {}.{}",
+                analyzer.getOperationType(),
+                table.getDatabaseName(),
+                table.getTableId());
+            created++;
+          } catch (RuntimeException e) {
+            // One bad table should not abort the rest of the database. Log and continue; the next
+            // analyzer pass will retry for any table whose save failed here.
+            log.error(
+                "Failed to create PENDING {} operation for table {}.{}: {}",
+                analyzer.getOperationType(),
+                table.getDatabaseName(),
+                table.getTableId(),
+                e.toString(),
+                e);
+            failed++;
+          }
+        }
+        log.info(
+            "Finished analyzing Database {}: created {} PENDING {} operation(s) ({} failed)",
+            databaseName,
+            created,
+            analyzer.getOperationType(),
+            failed);
+      }
+    }

...va/com/linkedin/openhouse/optimizer/analyzer/CadenceBasedOrphanFilesDeletionAnalyzer.java

-Original file line number
+Diff line change
@@ -0,0 +1,83 @@
+    package com.linkedin.openhouse.optimizer.analyzer;
+    import com.linkedin.openhouse.optimizer.model.OperationTypeDto;
+    import com.linkedin.openhouse.optimizer.model.TableDto;
+    import com.linkedin.openhouse.optimizer.model.TableOperationDto;
+    import com.linkedin.openhouse.optimizer.model.TableOperationsHistoryDto;
+    import java.time.Duration;
+    import java.util.Optional;
+    import org.springframework.beans.factory.annotation.Value;
+    import org.springframework.stereotype.Component;
+    /**
+     * Decides when to schedule an Orphan-Files-Deletion (OFD) run for a table.
+     *
+     * <p>OFD removes data files in the table's storage directory that are no longer referenced by any
+     * Iceberg snapshot — left-over output from failed writes, expired snapshots, or interrupted
+     * compactions. Running it too often wastes compute; running it too rarely lets orphan files
+     * accumulate and bloats storage cost. This analyzer balances the two on a per-table cadence.
+     *
+     * <h2>When OFD fires for a table</h2>
+     *
+     * All of the following must be true:
+     *
+     * <ol>
+     *   <li><b>Opt-in.</b> The table sets {@code maintenance.optimizer.ofd.enabled=true} in its table
+     *       properties. Without this flag, the analyzer ignores the table entirely.
+     *   <li><b>No active operation already in flight.</b> If the table has a non-CANCELED operation row
+     *       (PENDING, SCHEDULING, or SCHEDULED), the scheduler already owns it and the analyzer stays
+     *       out. A CANCELED row does not block — it is treated as if no operation exists.
+     *   <li><b>Cadence elapsed since the last completed run.</b>
+     *       <ul>
+     *         <li>If the table has <i>no</i> prior history, schedule immediately.
+     *         <li>If the most recent history entry is {@code SUCCESS}, wait {@code
+     *             ofd.success-retry-hours} (default 16h) after its {@code completedAt} before
+     *             scheduling again. Set below 24h so that even when a run lands at an unlucky time of
+     *             day, at least one re-evaluation is guaranteed within any rolling 24-hour window.
+     *         <li>If the most recent history entry is {@code FAILED}, wait {@code
+     *             ofd.failure-retry-hours} (default 1h) before retrying — shorter than the success
+     *             interval so transient failures recover quickly.
+     *       </ul>
+     * </ol>
+     *
+     * <p>The two retry intervals are configurable via {@code application.properties} and can be tuned
+     * per environment. The opt-in property is per-table and managed through the standard table-
+     * properties API.
+     */
+    @Component
+    public class CadenceBasedOrphanFilesDeletionAnalyzer implements OperationAnalyzer {
+      static final String OFD_ENABLED_PROPERTY = "maintenance.optimizer.ofd.enabled";
+      private final CadencePolicy cadencePolicy;
+      public CadenceBasedOrphanFilesDeletionAnalyzer(
+          @Value("${ofd.success-retry-hours:16}") long successRetryHours,
+          @Value("${ofd.failure-retry-hours:1}") long failureRetryHours) {
+        this.cadencePolicy =
+            new CadencePolicy(Duration.ofHours(successRetryHours), Duration.ofHours(failureRetryHours));
+      }
+      /** Package-private for tests that supply a pre-built {@link CadencePolicy}. */
+      CadenceBasedOrphanFilesDeletionAnalyzer(CadencePolicy cadencePolicy) {
+        this.cadencePolicy = cadencePolicy;
+      }
+      @Override
+      public OperationTypeDto getOperationType() {
+        return OperationTypeDto.ORPHAN_FILES_DELETION;
+      }
+      @Override
+      public boolean isEnabled(TableDto table) {
+        return "true".equals(table.getTableProperties().get(OFD_ENABLED_PROPERTY));
+      }
+      @Override
+      public boolean shouldSchedule(
+          TableDto table,
+          Optional<TableOperationDto> currentOp,
+          Optional<TableOperationsHistoryDto> latestHistory) {
+        return cadencePolicy.shouldSchedule(currentOp, latestHistory);
+      }
+    }

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

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

Uh oh!

Diff view

Diff view

There are no files selected for viewing

Uh oh!

Uh oh!

Uh oh!