[BDP-102028] feat(optimizer): [1/N] Optimizer Database

build.gradle

@@ -177,6 +177,7 @@ tasks.register('CopyGitHooksTask', Copy) {
 //   tables-service.Dockerfile      -> :services:tables:bootJar
 //   housetables-service.Dockerfile -> :services:housetables:bootJar
 //   jobs-service.Dockerfile        -> :services:jobs:bootJar
+//   optimizer-service.Dockerfile   -> :services:optimizer:bootJar
 //   jobs-scheduler.Dockerfile      -> :apps:openhouse-spark-apps_2.12:shadowJar (uber JAR)
 //   spark-base-hadoop2.8.dockerfile ->
 //       :integrations:spark:spark-3.1:openhouse-spark-runtime_2.12:shadowJar (uber JAR)
@@ -196,6 +197,7 @@ tasks.register('dockerPrereqs') {
   dependsOn ':services:tables:bootJar'
   dependsOn ':services:housetables:bootJar'
   dependsOn ':services:jobs:bootJar'
+  dependsOn ':services:optimizer:bootJar'
 
   // Spark runtime uber JARs (shadowJar)
   dependsOn ':integrations:spark:spark-3.1:openhouse-spark-runtime_2.12:shadowJar'
@@ -219,6 +221,7 @@ tasks.register('dockerPrereqs') {
     println '  build/tables/libs/tables.jar'
     println '  build/housetables/libs/housetables.jar'
     println '  build/jobs/libs/jobs.jar'
+    println '  build/optimizer/libs/optimizer.jar'
     println '  build/openhouse-spark-runtime_2.12/libs/openhouse-spark-runtime_2.12-uber.jar'
     println '  build/openhouse-spark-3.5-runtime_2.12/libs/openhouse-spark-3.5-runtime_2.12-uber.jar'
     println '  build/openhouse-spark-apps_2.12/libs/openhouse-spark-apps_2.12-uber.jar'

services/optimizer/build.gradle

@@ -0,0 +1,17 @@
+plugins {
+  id 'openhouse.springboot-ext-conventions'
+  id 'org.springframework.boot' version '2.7.8'
+}
+
+dependencies {
+  implementation 'org.springframework.boot:spring-boot-starter-data-jpa:2.7.8'
+  implementation 'com.vladmihalcea:hibernate-types-55:2.21.1'
+  implementation 'org.springframework.boot:spring-boot-starter-web:2.7.8'
+  implementation 'mysql:mysql-connector-java:8.+'
+  testImplementation 'com.h2database:h2:2.2.224'
+  testImplementation 'org.springframework.boot:spring-boot-starter-test:2.7.8'
+}
+
+test {
+  useJUnitPlatform()
+}

services/optimizer/src/main/java/com/linkedin/openhouse/optimizer/OptimizerServiceApplication.java

@@ -0,0 +1,13 @@
+package com.linkedin.openhouse.optimizer;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+/** Spring Boot entry point for the Optimizer Service. */
+@SpringBootApplication
+public class OptimizerServiceApplication {
+
+  public static void main(String[] args) {
+    SpringApplication.run(OptimizerServiceApplication.class, args);
+  }
+}

services/optimizer/src/main/java/com/linkedin/openhouse/optimizer/api/model/CompleteOperationRequest.java

@@ -0,0 +1,47 @@
+package com.linkedin.openhouse.optimizer.api.model;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * Request body for {@code POST /v1/table-operations/complete}.
+ *
+ * <p>Reports the outcome of a single completed operation. The service looks up the operation row by
+ * {@link #operationId} and writes a history entry for it.
+ *
+ * <p>A single Spark job typically processes N tables and yields N independent (status) outcomes —
+ * one per operation. Callers issue one complete request per operation; the service does not
+ * bulk-complete by job.
+ *
+ * <p>The remaining fields ({@link #tableUuid}, {@link #databaseName}, {@link #tableName}, {@link
+ * #operationType}) are debug-only echo information. The server does not key off them; they are
+ * preserved on log lines and traces so an operator looking at a failing complete call can see which
+ * (db, table, operation) the caller believed it was completing without joining back to the
+ * operation row.
+ */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class CompleteOperationRequest {
+
+  /** Operation row's UUID — the primary lookup key. */
+  private String operationId;
+
+  /** Terminal outcome for this single operation. */
+  private HistoryStatus status;
+
+  /** Debug echo: stable table identity the caller believed it was completing. */
+  private String tableUuid;
+
+  /** Debug echo: database name. */
+  private String databaseName;
+
+  /** Debug echo: table name. */
+  private String tableName;
+
+  /** Debug echo: operation type. */
+  private OperationType operationType;
+}

services/optimizer/src/main/java/com/linkedin/openhouse/optimizer/api/model/HistoryStatus.java

@@ -0,0 +1,21 @@
+package com.linkedin.openhouse.optimizer.api.model;
+
+/** Terminal states for a completed Spark maintenance job. */
+public enum HistoryStatus {
+
+  /** The Spark job for this operation completed successfully. */
+  SUCCESS,
+
+  /** The Spark job for this operation failed. */
+  FAILED;
+
+  /** Convert to the internal-model counterpart. */
+  public com.linkedin.openhouse.optimizer.model.HistoryStatus toModel() {
+    return com.linkedin.openhouse.optimizer.model.HistoryStatus.valueOf(name());
+  }
+
+  /** Build the api-layer enum from the internal-model counterpart. */
+  public static HistoryStatus fromModel(com.linkedin.openhouse.optimizer.model.HistoryStatus v) {
+    return v == null ? null : HistoryStatus.valueOf(v.name());
+  }
+}

services/optimizer/src/main/java/com/linkedin/openhouse/optimizer/api/model/OperationStatus.java

@@ -0,0 +1,32 @@
+package com.linkedin.openhouse.optimizer.api.model;
+
+/** Lifecycle states for a table operation recommendation. */
+public enum OperationStatus {
+
+  /** Recommended by the Analyzer but not yet claimed by the Scheduler. */
+  PENDING,
+
+  /** Claimed by the Scheduler; waiting for the Jobs Service to return a job ID. */
+  SCHEDULING,
+
+  /** Job submitted to the Jobs Service; the row now carries a {@code jobId}. */
+  SCHEDULED,
+
+  /**
+   * Marked by the Scheduler when it detects duplicate PENDING rows for the same {@code (table_uuid,
+   * operation_type)}. Only the most-recent PENDING row is claimed; older duplicates are CANCELED
+   * before the claim step.
+   */
+  CANCELED;
+
+  /** Convert to the internal-model counterpart. */
+  public com.linkedin.openhouse.optimizer.model.OperationStatus toModel() {
+    return com.linkedin.openhouse.optimizer.model.OperationStatus.valueOf(name());
+  }
+
+  /** Build the api-layer enum from the internal-model counterpart. */
+  public static OperationStatus fromModel(
+      com.linkedin.openhouse.optimizer.model.OperationStatus v) {
+    return v == null ? null : OperationStatus.valueOf(v.name());
+  }
+}

services/optimizer/src/main/java/com/linkedin/openhouse/optimizer/api/model/OperationType.java

@@ -0,0 +1,17 @@
+package com.linkedin.openhouse.optimizer.api.model;
+
+/** Maintenance operation types supported by the continuous optimizer. */
+public enum OperationType {
+  /** Removes orphaned data files no longer referenced by table metadata. */
+  ORPHAN_FILES_DELETION;
+
+  /** Convert to the internal-model counterpart. */
+  public com.linkedin.openhouse.optimizer.model.OperationType toModel() {
+    return com.linkedin.openhouse.optimizer.model.OperationType.valueOf(name());
+  }
+
+  /** Build the api-layer enum from the internal-model counterpart. */
+  public static OperationType fromModel(com.linkedin.openhouse.optimizer.model.OperationType v) {
+    return v == null ? null : OperationType.valueOf(v.name());
+  }
+}

services/optimizer/src/main/java/com/linkedin/openhouse/optimizer/api/model/TableOperationsDto.java

@@ -0,0 +1,74 @@
+package com.linkedin.openhouse.optimizer.api.model;
+
+import com.linkedin.openhouse.optimizer.model.TableOperation;
+import java.time.Instant;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/** DTO for {@code table_operations} — Analyzer recommendations read by the Scheduler. */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class TableOperationsDto {
+
+  /** Client-generated UUID identifying this specific operation recommendation. */
+  private String id;
+
+  /** Stable table identity from the Tables Service. */
+  private String tableUuid;
+
+  /** Denormalized database name for display; not part of the primary key. */
+  private String databaseName;
+
+  /** Denormalized table name for display; not part of the primary key. */
+  private String tableName;
+
+  /** The type of maintenance operation (e.g. ORPHAN_FILES_DELETION). */
+  private OperationType operationType;
+
+  /** {@code PENDING} or {@code SCHEDULED}. Defaults to {@code PENDING} on creation. */
+  private OperationStatus status;
+
+  /** Server-set when the row is first created by the Analyzer. */
+  private Instant createdAt;
+
+  /** Set by the Scheduler when claiming; {@code null} while PENDING. */
+  private Instant scheduledAt;
+
+  /** Job ID returned by the Jobs Service after successful submission. */
+  private String jobId;
+
+  /** Convert to the internal-model counterpart. */
+  public TableOperation toModel() {
+    return TableOperation.builder()
+        .id(id)
+        .tableUuid(tableUuid)
+        .databaseName(databaseName)
+        .tableName(tableName)
+        .operationType(operationType == null ? null : operationType.toModel())
+        .status(status == null ? null : status.toModel())
+        .createdAt(createdAt)
+        .scheduledAt(scheduledAt)
+        .build();
+  }
+
+  /** Build a wire DTO from the internal-model counterpart. */
+  public static TableOperationsDto fromModel(TableOperation op) {
+    if (op == null) {
+      return null;
+    }
+    return TableOperationsDto.builder()
+        .id(op.getId())
+        .tableUuid(op.getTableUuid())
+        .databaseName(op.getDatabaseName())
+        .tableName(op.getTableName())
+        .operationType(OperationType.fromModel(op.getOperationType()))
+        .status(OperationStatus.fromModel(op.getStatus()))
+        .createdAt(op.getCreatedAt())
+        .scheduledAt(op.getScheduledAt())
+        .build();
+  }
+}

services/optimizer/src/main/java/com/linkedin/openhouse/optimizer/api/model/TableOperationsHistoryDto.java

@@ -0,0 +1,66 @@
+package com.linkedin.openhouse.optimizer.api.model;
+
+import com.linkedin.openhouse.optimizer.model.TableOperationsHistory;
+import java.time.Instant;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/** DTO for {@code table_operations_history} — append-only operation results. */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class TableOperationsHistoryDto {
+
+  /** Same UUID as the originating {@code table_operations.id}; supplied by the caller. */
+  private String id;
+
+  /** Stable table identity from the Tables Service. */
+  private String tableUuid;
+
+  /** Denormalized database name for display. */
+  private String databaseName;
+
+  /** Denormalized table name for display. */
+  private String tableName;
+
+  /** The type of maintenance operation this history row records. */
+  private OperationType operationType;
+
+  /** When the operation completed, as recorded by the complete endpoint. */
+  private Instant completedAt;
+
+  /** {@code SUCCESS} or {@code FAILED}. */
+  private HistoryStatus status;
+
+  /** Convert to the internal-model counterpart. */
+  public TableOperationsHistory toModel() {
+    return TableOperationsHistory.builder()
+        .id(id)
+        .tableUuid(tableUuid)
+        .databaseName(databaseName)
+        .tableName(tableName)
+        .operationType(operationType == null ? null : operationType.toModel())
+        .completedAt(completedAt)
+        .status(status == null ? null : status.toModel())
+        .build();
+  }
+
+  /** Build a wire DTO from the internal-model counterpart. */
+  public static TableOperationsHistoryDto fromModel(TableOperationsHistory h) {
+    if (h == null) {
+      return null;
+    }
+    return TableOperationsHistoryDto.builder()
+        .id(h.getId())
+        .tableUuid(h.getTableUuid())
+        .databaseName(h.getDatabaseName())
+        .tableName(h.getTableName())
+        .operationType(OperationType.fromModel(h.getOperationType()))
+        .completedAt(h.getCompletedAt())
+        .status(HistoryStatus.fromModel(h.getStatus()))
+        .build();
+  }
+}