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

services/optimizer/build.gradle

@@ -4,7 +4,11 @@ plugins {
 }
 
 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'
 }
 

services/optimizer/src/main/java/com/linkedin/openhouse/optimizer/db/CommitDeltaMetrics.java

@@ -0,0 +1,28 @@
+package com.linkedin.openhouse.optimizer.db;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/** Per-commit incremental counters. Serialized as JSON into the {@code delta} column. */
+@Data
+@Builder(toBuilder = true)
+@NoArgsConstructor
+@AllArgsConstructor
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class CommitDeltaMetrics {
+
+  /** Number of data files this commit added to the table. */
+  private Long numFilesAdded;
+
+  /** Number of data files this commit removed from the table. */
+  private Long numFilesDeleted;
+
+  /** Total bytes added by this commit. */
+  private Long addedSizeBytes;
+
+  /** Total bytes removed by this commit. */
+  private Long deletedSizeBytes;
+}

services/optimizer/src/main/java/com/linkedin/openhouse/optimizer/db/HistoryStatus.java

@@ -0,0 +1,15 @@
+package com.linkedin.openhouse.optimizer.db;
+
+/**
+ * DB-layer enum for the {@code status} column of {@code table_operations_history}.
+ *
+ * <p>Self-contained: no references to api/ or model/ types.
+ */
+public enum HistoryStatus {
+
+  /** The Spark job for this operation completed successfully. */
+  SUCCESS,
+
+  /** The Spark job for this operation failed. */
+  FAILED
+}

services/optimizer/src/main/java/com/linkedin/openhouse/optimizer/db/OperationStatus.java

@@ -0,0 +1,21 @@
+package com.linkedin.openhouse.optimizer.db;
+
+/**
+ * DB-layer enum for the {@code status} column of {@code table_operations}.
+ *
+ * <p>Self-contained: no references to api/ or model/ types.
+ */
+public enum OperationStatus {
+
+  /** Analyzer has written the row; not yet claimed by the scheduler. */
+  PENDING,
+
+  /** Scheduler has claimed the row and is launching a job; jobId not yet recorded. */
+  SCHEDULING,
+
+  /** Job has been submitted to the Jobs Service; the row carries a {@code jobId}. */
+  SCHEDULED,
+
+  /** Scheduler marked this row as a duplicate of another PENDING row; not claimable. */
+  CANCELED
+}

services/optimizer/src/main/java/com/linkedin/openhouse/optimizer/db/OperationType.java

@@ -0,0 +1,14 @@
+package com.linkedin.openhouse.optimizer.db;
+
+/**
+ * DB-layer enum for the operation types persisted in {@code table_operations.operation_type} and
+ * {@code table_operations_history.operation_type}.
+ *
+ * <p>Self-contained: no references to api/ or model/ types. JPA binds this via
+ * {@code @Enumerated(EnumType.STRING)}.
+ */
+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/db/SnapshotMetrics.java

@@ -0,0 +1,28 @@
+package com.linkedin.openhouse.optimizer.db;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/** Point-in-time snapshot fields. Serialized as JSON into the {@code snapshot} column. */
+@Data
+@Builder(toBuilder = true)
+@NoArgsConstructor
+@AllArgsConstructor
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class SnapshotMetrics {
+
+  /** Iceberg metadata version pointer for this snapshot. */
+  private String tableVersion;
+
+  /** Filesystem path (or URI) of the table's storage root. */
+  private String tableLocation;
+
+  /** Total on-disk size of the table at this snapshot, in bytes. */
+  private Long tableSizeBytes;
+
+  /** Total number of data files as of the latest snapshot — used for bin-packing. */
+  private Long numCurrentFiles;
+}

services/optimizer/src/main/java/com/linkedin/openhouse/optimizer/db/TableOperationsHistoryRow.java

@@ -0,0 +1,75 @@
+package com.linkedin.openhouse.optimizer.db;
+
+import java.time.Instant;
+import javax.persistence.Column;
+import javax.persistence.Entity;
+import javax.persistence.EnumType;
+import javax.persistence.Enumerated;
+import javax.persistence.Id;
+import javax.persistence.Index;
+import javax.persistence.Table;
+import lombok.AccessLevel;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+
+/**
+ * Append-only record of a completed maintenance operation.
+ *
+ * <p>Written when the operation-complete endpoint is called. The {@code id} is the same UUID as the
+ * originating live-operations row, tying each history entry back to the operation cycle that
+ * produced it. Multiple runs of the same operation on the same table produce multiple rows.
+ *
+ * <p>Self-contained DB-layer type: enums are {@link OperationType} / {@link HistoryStatus} from the
+ * same package, JPA-bound as strings.
+ */
+@Entity
+@Table(
+    name = "table_operations_history",
+    indexes = {
+      @Index(name = "idx_table_uuid_hist", columnList = "table_uuid"),
+      @Index(name = "idx_op_type_hist", columnList = "operation_type"),
+      @Index(name = "idx_completed_at", columnList = "completed_at"),
+      @Index(name = "idx_status_hist", columnList = "status"),
+      @Index(name = "idx_toph_db_table", columnList = "database_name, table_name")
+    })
+@Getter
+@EqualsAndHashCode
+@Builder(toBuilder = true)
+@NoArgsConstructor(access = AccessLevel.PROTECTED)
+@AllArgsConstructor(access = AccessLevel.PROTECTED)
+public class TableOperationsHistoryRow {
+
+  /** Same UUID as the originating live-operations row. Set by the caller; not generated. */
+  @Id
+  @Column(name = "id", nullable = false, length = 36)
+  private String id;
+
+  /** Stable table identity from the Tables Service. */
+  @Column(name = "table_uuid", nullable = false, length = 36)
+  private String tableUuid;
+
+  /** Denormalized database name. */
+  @Column(name = "database_name", nullable = false, length = 128)
+  private String databaseName;
+
+  /** Denormalized table name. */
+  @Column(name = "table_name", nullable = false, length = 128)
+  private String tableName;
+
+  /** The type of maintenance operation this history row records. */
+  @Enumerated(EnumType.STRING)
+  @Column(name = "operation_type", nullable = false, length = 50)
+  private OperationType operationType;
+
+  /** When the operation completed, as recorded by the complete endpoint. */
+  @Column(name = "completed_at", nullable = false)
+  private Instant completedAt;
+
+  /** Terminal outcome: {@link HistoryStatus#SUCCESS} or {@link HistoryStatus#FAILED}. */
+  @Enumerated(EnumType.STRING)
+  @Column(name = "status", nullable = false, length = 20)
+  private HistoryStatus status;
+}

services/optimizer/src/main/java/com/linkedin/openhouse/optimizer/db/TableOperationsRow.java

@@ -0,0 +1,85 @@
+package com.linkedin.openhouse.optimizer.db;
+
+import java.time.Instant;
+import javax.persistence.Column;
+import javax.persistence.Entity;
+import javax.persistence.EnumType;
+import javax.persistence.Enumerated;
+import javax.persistence.Id;
+import javax.persistence.Index;
+import javax.persistence.Table;
+import lombok.AccessLevel;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+
+/**
+ * JPA entity representing an Analyzer recommendation for a table maintenance operation.
+ *
+ * <p>Each row is identified by a client-generated UUID ({@code id}). The Analyzer creates a new row
+ * when it first recommends an operation for a table, or when re-recommending after a prior terminal
+ * state. {@code table_uuid} is the stable identity for the table (survives renames; rotates on
+ * drop+recreate). The application enforces one active (PENDING / SCHEDULING / SCHEDULED) row per
+ * {@code (table_uuid, operation_type)} at a time.
+ *
+ * <p>Self-contained DB-layer type: enums are {@link OperationType} / {@link OperationStatus} from
+ * the same package, JPA-bound as strings.
+ */
+@Entity
+@Table(
+    name = "table_operations",
+    indexes = {
+      @Index(name = "idx_table_uuid", columnList = "table_uuid"),
+      @Index(name = "idx_op_type", columnList = "operation_type"),
+      @Index(name = "idx_status", columnList = "status"),
+      @Index(name = "idx_created_at", columnList = "created_at"),
+      @Index(name = "idx_scheduled_at", columnList = "scheduled_at")
+    })
+@Getter
+@EqualsAndHashCode
+@Builder(toBuilder = true)
+@NoArgsConstructor(access = AccessLevel.PROTECTED)
+@AllArgsConstructor(access = AccessLevel.PROTECTED)
+public class TableOperationsRow {
+
+  /** Client-generated UUID identifying this specific operation recommendation. */
+  @Id
+  @Column(name = "id", nullable = false, length = 36)
+  private String id;
+
+  /** Stable table identity from the Tables Service. Survives renames; rotates on drop+recreate. */
+  @Column(name = "table_uuid", nullable = false, length = 36)
+  private String tableUuid;
+
+  /** Denormalized database name. */
+  @Column(name = "database_name", nullable = false, length = 128)
+  private String databaseName;
+
+  /** Denormalized table name. */
+  @Column(name = "table_name", nullable = false, length = 128)
+  private String tableName;
+
+  /** The type of maintenance operation this row recommends. */
+  @Enumerated(EnumType.STRING)
+  @Column(name = "operation_type", nullable = false, length = 50)
+  private OperationType operationType;
+
+  /** Lifecycle state — drives the scheduler's CAS claim and the analyzer's eligibility check. */
+  @Enumerated(EnumType.STRING)
+  @Column(name = "status", nullable = false, length = 20)
+  private OperationStatus status;
+
+  /** When the analyzer first created this row. Set on insert; never updated. */
+  @Column(name = "created_at", nullable = false)
+  private Instant createdAt;
+
+  /** When the scheduler last submitted a job for this row. {@code null} while {@code PENDING}. */
+  @Column(name = "scheduled_at")
+  private Instant scheduledAt;
+
+  /** Spark job ID written by the scheduler at claim time. Internal-only; never exposed on wire. */
+  @Column(name = "job_id", length = 255)
+  private String jobId;
+}

services/optimizer/src/main/java/com/linkedin/openhouse/optimizer/db/TableStatsHistoryRow.java

@@ -0,0 +1,74 @@
+package com.linkedin.openhouse.optimizer.db;
+
+import com.vladmihalcea.hibernate.type.json.JsonStringType;
+import java.time.Instant;
+import javax.persistence.Column;
+import javax.persistence.Entity;
+import javax.persistence.Id;
+import javax.persistence.Index;
+import javax.persistence.Table;
+import lombok.AccessLevel;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import org.hibernate.annotations.Type;
+import org.hibernate.annotations.TypeDef;
+
+/**
+ * Append-only record of per-commit stats reported by the Tables Service.
+ *
+ * <p>Each Iceberg commit produces one row. Consumers can query this table to reconstruct change
+ * rates over arbitrary time windows.
+ *
+ * <p>Self-contained DB-layer type. The stats payload is split across two JSON columns — {@link
+ * SnapshotMetrics} (point-in-time fields at commit time) and {@link CommitDeltaMetrics} (per-commit
+ * counters).
+ */
+@TypeDef(name = "json", typeClass = JsonStringType.class)
+@Entity
+@Table(
+    name = "table_stats_history",
+    indexes = {
+      @Index(name = "idx_tsh_table_uuid", columnList = "table_uuid"),
+      @Index(name = "idx_tsh_recorded_at", columnList = "recorded_at")
+    })
+@Getter
+@EqualsAndHashCode
+@Builder(toBuilder = true)
+@NoArgsConstructor(access = AccessLevel.PROTECTED)
+@AllArgsConstructor(access = AccessLevel.PROTECTED)
+public class TableStatsHistoryRow {
+
+  /** UUID primary key — set by the caller, not generated server-side. */
+  @Id
+  @Column(name = "id", nullable = false, length = 36)
+  private String id;
+
+  /** Stable Iceberg table UUID. */
+  @Column(name = "table_uuid", nullable = false, length = 36)
+  private String tableUuid;
+
+  /** Denormalized database name. */
+  @Column(name = "database_name", nullable = false, length = 128)
+  private String databaseName;
+
+  /** Denormalized table name. */
+  @Column(name = "table_name", nullable = false, length = 128)
+  private String tableName;
+
+  /** Snapshot fields at commit time. Stored as a JSON blob in the {@code snapshot} column. */
+  @Type(type = "json")
+  @Column(name = "snapshot", columnDefinition = "TEXT")
+  private SnapshotMetrics snapshot;
+
+  /** Per-commit delta counters. Stored as a JSON blob in the {@code delta} column. */
+  @Type(type = "json")
+  @Column(name = "delta", columnDefinition = "TEXT")
+  private CommitDeltaMetrics delta;
+
+  /** When this history row was recorded (commit time). */
+  @Column(name = "recorded_at", nullable = false)
+  private Instant recordedAt;
+}