feat(optimizer): [0/N] Optimizer Data Model

build.gradle

@@ -157,6 +157,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)
@@ -176,6 +177,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'
@@ -196,6 +198,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/mapper/OptimizerMapper.java

@@ -0,0 +1,32 @@
+package com.linkedin.openhouse.optimizer.api.mapper;
+
+import com.linkedin.openhouse.optimizer.api.model.TableOperationsDto;
+import com.linkedin.openhouse.optimizer.api.model.TableOperationsHistoryDto;
+import com.linkedin.openhouse.optimizer.api.model.TableStatsDto;
+import com.linkedin.openhouse.optimizer.api.model.TableStatsHistoryDto;
+import com.linkedin.openhouse.optimizer.entity.TableOperationsHistoryRow;
+import com.linkedin.openhouse.optimizer.entity.TableOperationsRow;
+import com.linkedin.openhouse.optimizer.entity.TableStatsHistoryRow;
+import com.linkedin.openhouse.optimizer.entity.TableStatsRow;
+import org.mapstruct.Mapper;
+
+/**
+ * MapStruct mapper for converting between optimizer JPA entities and their corresponding DTOs.
+ *
+ * <p>Spring-instantiated at compile time. Inject via {@code @Autowired} or constructor injection.
+ */
+@Mapper(componentModel = "spring")
+public interface OptimizerMapper {
+
+  /** Map a {@link TableOperationsRow} to its DTO. */
+  TableOperationsDto toDto(TableOperationsRow row);
+
+  /** Map a {@link TableOperationsHistoryRow} to its DTO. */
+  TableOperationsHistoryDto toDto(TableOperationsHistoryRow row);
+
+  /** Map a {@link TableStatsRow} to its DTO. */
+  TableStatsDto toDto(TableStatsRow row);
+
+  /** Map a {@link TableStatsHistoryRow} to its DTO. */
+  TableStatsHistoryDto toDto(TableStatsHistoryRow row);
+}

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

@@ -0,0 +1,25 @@
+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/{id}/complete}.
+ *
+ * <p>Reports the outcome of a completed operation. The backend looks up the operation row by {@code
+ * id} and writes a history entry with the operation's table metadata and the supplied result.
+ */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class CompleteOperationRequest {
+
+  /** Outcome of the operation. */
+  private OperationHistoryStatus status;
+
+  /** Error details on failure; {@code null} on success. */
+  private JobResult result;
+}

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

@@ -0,0 +1,25 @@
+package com.linkedin.openhouse.optimizer.api.model;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * Result payload for a completed Spark maintenance job.
+ *
+ * <p>Stored as JSON in the {@code result} column of {@code table_operations_history}. Both fields
+ * are {@code null} on success; populated on failure.
+ */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class JobResult {
+
+  /** Human-readable error message; {@code null} if the job succeeded. */
+  private String errorMessage;
+
+  /** Error category (e.g., {@code OOM}, {@code TIMEOUT}); {@code null} if the job succeeded. */
+  private String errorType;
+}

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

@@ -0,0 +1,7 @@
+package com.linkedin.openhouse.optimizer.api.model;
+
+/** Terminal states for a completed Spark maintenance job. */
+public enum OperationHistoryStatus {
+  SUCCESS,
+  FAILED
+}

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

@@ -0,0 +1,21 @@
+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
+}

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

@@ -0,0 +1,7 @@
+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
+}

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

@@ -0,0 +1,45 @@
+package com.linkedin.openhouse.optimizer.api.model;
+
+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;
+
+  /** Reserved for future per-operation metadata; currently unused. */
+  private String metrics;
+}

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

@@ -0,0 +1,37 @@
+package com.linkedin.openhouse.optimizer.api.model;
+
+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;
+
+  private String databaseName;
+  private String tableName;
+  private OperationType operationType;
+
+  /** When the operation completed, as recorded by the complete endpoint. */
+  private Instant submittedAt;
+
+  /** {@code SUCCESS} or {@code FAILED}. */
+  private OperationHistoryStatus status;
+
+  /** Job ID from the Jobs Service. */
+  private String jobId;
+
+  /** Job result payload; both fields null on success. */
+  private JobResult result;
+}

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

@@ -0,0 +1,49 @@
+package com.linkedin.openhouse.optimizer.api.model;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/** Combined stats payload stored as a single JSON blob per table. */
+@Data
+@Builder(toBuilder = true)
+@NoArgsConstructor
+@AllArgsConstructor
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class TableStats {
+
+  /** Snapshot fields — overwritten on every upsert. */
+  private SnapshotMetrics snapshot;
+
+  /** Delta fields — accumulated across commit events. */
+  private CommitDelta delta;
+
+  /** Point-in-time metadata read from Iceberg at scan time. */
+  @Data
+  @Builder(toBuilder = true)
+  @NoArgsConstructor
+  @AllArgsConstructor
+  @JsonIgnoreProperties(ignoreUnknown = true)
+  public static class SnapshotMetrics {
+    private String clusterId;
+    private String tableVersion;
+    private String tableLocation;
+    private Long tableSizeBytes;
+    /** Total number of data files as of the latest snapshot — used for bin-packing. */
+    private Long numCurrentFiles;
+  }
+
+  /** Per-commit incremental counters; accumulated across all recorded commit events. */
+  @Data
+  @Builder(toBuilder = true)
+  @NoArgsConstructor
+  @AllArgsConstructor
+  public static class CommitDelta {
+    private Long numFilesAdded;
+    private Long numFilesDeleted;
+    private Long addedSizeBytes;
+    private Long deletedSizeBytes;
+  }
+}

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

@@ -0,0 +1,34 @@
+package com.linkedin.openhouse.optimizer.api.model;
+
+import java.time.Instant;
+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 TableStatsDto {
+
+  /** Stable Iceberg table UUID. Primary key of the stats row. */
+  private String tableUuid;
+
+  /** Denormalized database name for display. */
+  private String databaseId;
+
+  /** Denormalized table name for display. */
+  private String tableName;
+
+  /** Combined snapshot + delta stats payload, stored as JSON. */
+  private TableStats 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;
+}

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

@@ -0,0 +1,33 @@
+package com.linkedin.openhouse.optimizer.api.model;
+
+import java.time.Instant;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/** DTO for {@code table_stats_history} — used for response payloads. */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class TableStatsHistoryDto {
+
+  /** Auto-increment primary key. */
+  private Long id;
+
+  /** Stable Iceberg table UUID. */
+  private String tableUuid;
+
+  /** Denormalized database name for display. */
+  private String databaseId;
+
+  /** Denormalized table name for display. */
+  private String tableName;
+
+  /** Snapshot + delta stats from this commit event. */
+  private TableStats stats;
+
+  /** When this history row was recorded. */
+  private Instant recordedAt;
+}