[BDP-102028] feat(optimizer): [0/N] Optimizer API and internal model

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,13 @@
+plugins {
+  id 'openhouse.springboot-ext-conventions'
+  id 'org.springframework.boot' version '2.7.8'
+}
+
+dependencies {
+  implementation 'org.springframework.boot:spring-boot-starter-web:2.7.8'
+  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/spec/HistoryStatus.java

@@ -0,0 +1,21 @@
+package com.linkedin.openhouse.optimizer.api.spec;
+
+/** 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.HistoryStatusDto toModel() {
+    return com.linkedin.openhouse.optimizer.model.HistoryStatusDto.valueOf(name());
+  }
+
+  /** Build the api-layer enum from the internal-model counterpart. */
+  public static HistoryStatus fromModel(com.linkedin.openhouse.optimizer.model.HistoryStatusDto v) {
+    return v == null ? null : HistoryStatus.valueOf(v.name());
+  }
+}

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

@@ -0,0 +1,32 @@
+package com.linkedin.openhouse.optimizer.api.spec;
+
+/** 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.OperationStatusDto toModel() {
+    return com.linkedin.openhouse.optimizer.model.OperationStatusDto.valueOf(name());
+  }
+
+  /** Build the api-layer enum from the internal-model counterpart. */
+  public static OperationStatus fromModel(
+      com.linkedin.openhouse.optimizer.model.OperationStatusDto v) {
+    return v == null ? null : OperationStatus.valueOf(v.name());
+  }
+}

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

@@ -0,0 +1,17 @@
+package com.linkedin.openhouse.optimizer.api.spec;
+
+/** 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.OperationTypeDto toModel() {
+    return com.linkedin.openhouse.optimizer.model.OperationTypeDto.valueOf(name());
+  }
+
+  /** Build the api-layer enum from the internal-model counterpart. */
+  public static OperationType fromModel(com.linkedin.openhouse.optimizer.model.OperationTypeDto v) {
+    return v == null ? null : OperationType.valueOf(v.name());
+  }
+}

services/optimizer/src/main/java/com/linkedin/openhouse/optimizer/api/spec/TableOperations.java

@@ -0,0 +1,76 @@
+package com.linkedin.openhouse.optimizer.api.spec;
+
+import com.linkedin.openhouse.optimizer.model.TableOperationDto;
+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 TableOperations {
+
+  /** 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 TableOperationDto toModel() {
+    return TableOperationDto.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)
+        .jobId(jobId)
+        .build();
+  }
+
+  /** Build a wire DTO from the internal-model counterpart. */
+  public static TableOperations fromModel(TableOperationDto op) {
+    if (op == null) {
+      return null;
+    }
+    return TableOperations.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())
+        .jobId(op.getJobId())
+        .build();
+  }
+}

services/optimizer/src/main/java/com/linkedin/openhouse/optimizer/api/spec/TableOperationsHistory.java

@@ -0,0 +1,66 @@
+package com.linkedin.openhouse.optimizer.api.spec;
+
+import com.linkedin.openhouse.optimizer.model.TableOperationsHistoryDto;
+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 TableOperationsHistory {
+
+  /** 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 TableOperationsHistoryDto toModel() {
+    return TableOperationsHistoryDto.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 TableOperationsHistory fromModel(TableOperationsHistoryDto h) {
+    if (h == null) {
+      return null;
+    }
+    return TableOperationsHistory.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();
+  }
+}

services/optimizer/src/main/java/com/linkedin/openhouse/optimizer/api/spec/TableStats.java

@@ -0,0 +1,70 @@
+package com.linkedin.openhouse.optimizer.api.spec;
+
+import java.time.Instant;
+import java.util.Collections;
+import java.util.Map;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/** DTO for {@code table_stats} — used for response payloads. */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class TableStats {
+
+  /** Stable Iceberg table UUID. Primary key of the stats row. */
+  private String tableUuid;
+
+  /** Denormalized database name for display. */
+  private String databaseName;
+
+  /** Denormalized table name for display. */
+  private String tableName;
+
+  /** Combined snapshot + delta stats payload, stored as JSON. */
+  private TableStatsPayload stats;
+
+  /** Current table properties snapshot (e.g. maintenance opt-in flags). */
+  private Map<String, String> tableProperties;
+
+  /** When this row was last written. Used for staleness monitoring. */
+  private Instant updatedAt;
+
+  /** Convert to the internal-model counterpart. */
+  public com.linkedin.openhouse.optimizer.model.TableStatsDto toModel() {
+    com.linkedin.openhouse.optimizer.model.TableStatsDto payload =
+        stats == null
+            ? new com.linkedin.openhouse.optimizer.model.TableStatsDto()
+            : stats.toModel();
+    return payload
+        .toBuilder()
+        .tableUuid(tableUuid)
+        .databaseName(databaseName)
+        .tableName(tableName)
+        .tableProperties(tableProperties != null ? tableProperties : Collections.emptyMap())
+        .updatedAt(updatedAt)
+        .build();
+  }
+
+  /** Build a wire DTO from the internal-model counterpart. */
+  public static TableStats fromModel(com.linkedin.openhouse.optimizer.model.TableStatsDto m) {
+    if (m == null) {
+      return null;
+    }
+    return TableStats.builder()
+        .tableUuid(m.getTableUuid())
+        .databaseName(m.getDatabaseName())
+        .tableName(m.getTableName())
+        .stats(
+            TableStatsPayload.builder()
+                .snapshot(TableStatsPayload.SnapshotMetricsDto.fromModel(m.getSnapshot()))
+                .delta(TableStatsPayload.CommitDeltaDto.fromModel(m.getDelta()))
+                .build())
+        .tableProperties(m.getTableProperties())
+        .updatedAt(m.getUpdatedAt())
+        .build();
+  }
+}