feat(optimizer): [2/N] Optimizer REST service layer

services/optimizer/src/main/java/com/linkedin/openhouse/optimizer/api/controller/TableOperationsController.java

@@ -0,0 +1,66 @@
+package com.linkedin.openhouse.optimizer.api.controller;
+
+import com.linkedin.openhouse.optimizer.api.model.CompleteOperationRequest;
+import com.linkedin.openhouse.optimizer.api.model.OperationStatus;
+import com.linkedin.openhouse.optimizer.api.model.OperationType;
+import com.linkedin.openhouse.optimizer.api.model.TableOperationsDto;
+import com.linkedin.openhouse.optimizer.api.model.TableOperationsHistoryDto;
+import com.linkedin.openhouse.optimizer.service.OptimizerDataService;
+import java.util.List;
+import lombok.RequiredArgsConstructor;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+/** REST controller for {@code table_operations}. */
+@RestController
+@RequestMapping("/v1/table-operations")
+@RequiredArgsConstructor
+public class TableOperationsController {
+
+  private final OptimizerDataService service;
+
+  /**
+   * Report that an operation has completed. The backend looks up the operation row, writes a
+   * history entry with the operation's table metadata and the supplied result. Returns 201 Created
+   * with the history row, or 404 if the operation does not exist.
+   */
+  @PostMapping("/{id}/complete")
+  public ResponseEntity<TableOperationsHistoryDto> completeOperation(
+      @PathVariable String id, @RequestBody CompleteOperationRequest request) {
+    return service
+        .completeOperation(id, request)
+        .map(dto -> ResponseEntity.status(HttpStatus.CREATED).body(dto))
+        .orElse(ResponseEntity.notFound().build());
+  }
+
+  /** Fetch a single operation row by its ID, regardless of status. Returns 404 if not found. */
+  @GetMapping("/{id}")
+  public ResponseEntity<TableOperationsDto> getTableOperation(@PathVariable String id) {
+    return service
+        .getTableOperation(id)
+        .map(ResponseEntity::ok)
+        .orElse(ResponseEntity.notFound().build());
+  }
+
+  /**
+   * List operations matching the given filters. All parameters are optional — omit all to return
+   * every row.
+   */
+  @GetMapping
+  public ResponseEntity<List<TableOperationsDto>> listTableOperations(
+      @RequestParam(required = false) OperationType operationType,
+      @RequestParam(required = false) OperationStatus status,
+      @RequestParam(required = false) String databaseName,
+      @RequestParam(required = false) String tableName,
+      @RequestParam(required = false) String tableUuid) {
+    return ResponseEntity.ok(
+        service.listTableOperations(operationType, status, databaseName, tableName, tableUuid));
+  }
+}

services/optimizer/src/main/java/com/linkedin/openhouse/optimizer/api/controller/TableOperationsHistoryController.java

@@ -0,0 +1,60 @@
+package com.linkedin.openhouse.optimizer.api.controller;
+
+import com.linkedin.openhouse.optimizer.api.model.OperationHistoryStatus;
+import com.linkedin.openhouse.optimizer.api.model.OperationType;
+import com.linkedin.openhouse.optimizer.api.model.TableOperationsHistoryDto;
+import com.linkedin.openhouse.optimizer.service.OptimizerDataService;
+import java.time.Instant;
+import java.util.List;
+import lombok.RequiredArgsConstructor;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+/** REST controller for {@code table_operations_history}. */
+@RestController
+@RequestMapping("/v1/table-operations-history")
+@RequiredArgsConstructor
+public class TableOperationsHistoryController {
+
+  private final OptimizerDataService service;
+
+  /** Append a completed-job result. Called by the SparkJob after each run (success or failure). */
+  @PostMapping
+  public ResponseEntity<TableOperationsHistoryDto> appendHistory(
+      @RequestBody TableOperationsHistoryDto dto) {
+    return ResponseEntity.status(HttpStatus.CREATED).body(service.appendHistory(dto));
+  }
+
+  /** Return the most recent history for a table, newest first, up to {@code limit} rows. */
+  @GetMapping("/{tableUuid}")
+  public ResponseEntity<List<TableOperationsHistoryDto>> getHistory(
+      @PathVariable String tableUuid, @RequestParam(defaultValue = "100") int limit) {
+    return ResponseEntity.ok(service.getHistory(tableUuid, limit));
+  }
+
+  /**
+   * List history rows matching the given filters, ordered newest first. All parameters are optional
+   * — omit all to return every row up to {@code limit}.
+   */
+  @GetMapping
+  public ResponseEntity<List<TableOperationsHistoryDto>> listHistory(
+      @RequestParam(required = false) String databaseName,
+      @RequestParam(required = false) String tableName,
+      @RequestParam(required = false) String tableUuid,
+      @RequestParam(required = false) OperationType operationType,
+      @RequestParam(required = false) OperationHistoryStatus status,
+      @RequestParam(required = false) Instant since,
+      @RequestParam(required = false) Instant until,
+      @RequestParam(defaultValue = "100") int limit) {
+    return ResponseEntity.ok(
+        service.listHistory(
+            databaseName, tableName, tableUuid, operationType, status, since, until, limit));
+  }
+}

services/optimizer/src/main/java/com/linkedin/openhouse/optimizer/api/controller/TableStatsController.java

@@ -0,0 +1,69 @@
+package com.linkedin.openhouse.optimizer.api.controller;
+
+import com.linkedin.openhouse.optimizer.api.model.TableStatsDto;
+import com.linkedin.openhouse.optimizer.api.model.TableStatsHistoryDto;
+import com.linkedin.openhouse.optimizer.api.model.UpsertTableStatsRequest;
+import com.linkedin.openhouse.optimizer.service.OptimizerDataService;
+import java.time.Instant;
+import java.util.List;
+import lombok.RequiredArgsConstructor;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.PutMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+/** REST controller for managing per-table stats in the optimizer DB. */
+@RestController
+@RequestMapping("/v1/table-stats")
+@RequiredArgsConstructor
+public class TableStatsController {
+
+  private final OptimizerDataService service;
+
+  /**
+   * Create or overwrite the stats row for {@code tableUuid}. Called by the Tables Service on every
+   * Iceberg commit. Idempotent.
+   */
+  @PutMapping("/{tableUuid}")
+  public ResponseEntity<TableStatsDto> upsertTableStats(
+      @PathVariable String tableUuid, @RequestBody UpsertTableStatsRequest request) {
+    return ResponseEntity.ok(service.upsertTableStats(tableUuid, request));
+  }
+
+  /** Fetch the stats row for {@code tableUuid}. Returns 404 if no stats have been written yet. */
+  @GetMapping("/{tableUuid}")
+  public ResponseEntity<TableStatsDto> getTableStats(@PathVariable String tableUuid) {
+    return service
+        .getTableStats(tableUuid)
+        .map(ResponseEntity::ok)
+        .orElse(ResponseEntity.notFound().build());
+  }
+
+  /**
+   * List stats rows matching the given filters. All parameters are optional — omit all to return
+   * every row.
+   */
+  @GetMapping
+  public ResponseEntity<List<TableStatsDto>> listTableStats(
+      @RequestParam(required = false) String databaseId,
+      @RequestParam(required = false) String tableName,
+      @RequestParam(required = false) String tableUuid) {
+    return ResponseEntity.ok(service.listTableStats(databaseId, tableName, tableUuid));
+  }
+
+  /**
+   * Return per-commit stats history for {@code tableUuid}, newest first. Optionally filter by
+   * {@code since} (inclusive) and cap at {@code limit} rows.
+   */
+  @GetMapping("/{tableUuid}/history")
+  public ResponseEntity<List<TableStatsHistoryDto>> getStatsHistory(
+      @PathVariable String tableUuid,
+      @RequestParam(required = false) Instant since,
+      @RequestParam(defaultValue = "100") int limit) {
+    return ResponseEntity.ok(service.getStatsHistory(tableUuid, since, limit));
+  }
+}

services/optimizer/src/main/java/com/linkedin/openhouse/optimizer/service/OptimizerDataService.java

@@ -0,0 +1,98 @@
+package com.linkedin.openhouse.optimizer.service;
+
+import com.linkedin.openhouse.optimizer.api.model.CompleteOperationRequest;
+import com.linkedin.openhouse.optimizer.api.model.OperationHistoryStatus;
+import com.linkedin.openhouse.optimizer.api.model.OperationStatus;
+import com.linkedin.openhouse.optimizer.api.model.OperationType;
+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.api.model.UpsertTableStatsRequest;
+import java.time.Instant;
+import java.util.List;
+import java.util.Optional;
+
+/** Service interface for optimizer data operations. */
+public interface OptimizerDataService {
+
+  // --- TableOperations ---
+
+  /**
+   * List operations matching the given filters. Every parameter is optional — pass {@code null} to
+   * skip that filter. No filters returns all rows.
+   */
+  List<TableOperationsDto> listTableOperations(
+      OperationType operationType,
+      OperationStatus status,
+      String databaseName,
+      String tableName,
+      String tableUuid);
+
+  /**
+   * Complete an operation by writing a history entry. Looks up the operation row by {@code id},
+   * copies its table metadata into a new history row, and saves it. Returns the history DTO, or
+   * empty if the operation does not exist.
+   */
+  Optional<TableOperationsHistoryDto> completeOperation(
+      String id, CompleteOperationRequest request);
+
+  /**
+   * Return the operation row for {@code id} regardless of status, or empty if it does not exist.
+   * Used to poll a specific operation (e.g. waiting for SUCCESS after a Spark job completes).
+   */
+  Optional<TableOperationsDto> getTableOperation(String id);
+
+  // --- TableStats ---
+
+  /**
+   * Create or update the stats row for {@code tableUuid}. Fully idempotent: the same call
+   * overwrites the previous snapshot with the latest commit values.
+   */
+  TableStatsDto upsertTableStats(String tableUuid, UpsertTableStatsRequest request);
+
+  /** Return the stats row for {@code tableUuid}, or empty if none exists. */
+  Optional<TableStatsDto> getTableStats(String tableUuid);
+
+  /**
+   * List stats rows matching the given filters. Every parameter is optional — pass {@code null} to
+   * skip that filter. No filters returns all rows.
+   */
+  List<TableStatsDto> listTableStats(String databaseId, String tableName, String tableUuid);
+
+  /**
+   * Return per-commit stats history for {@code tableUuid}, newest first.
+   *
+   * @param tableUuid the stable table UUID
+   * @param since if non-null, only return rows recorded at or after this instant
+   * @param limit maximum number of rows to return
+   */
+  List<TableStatsHistoryDto> getStatsHistory(String tableUuid, Instant since, int limit);
+
+  // --- TableOperationsHistory ---
+
+  /** Append a completed-job result record. */
+  TableOperationsHistoryDto appendHistory(TableOperationsHistoryDto dto);
+
+  /**
+   * Return the most recent history rows for a table UUID, newest first.
+   *
+   * @param tableUuid the stable table UUID
+   * @param limit maximum number of rows to return
+   */
+  List<TableOperationsHistoryDto> getHistory(String tableUuid, int limit);
+
+  /**
+   * List history rows matching the given filters, ordered newest first. Every parameter is optional
+   * — pass {@code null} to skip that filter. No filters returns all rows up to {@code limit}.
+   */
+  List<TableOperationsHistoryDto> listHistory(
+      String databaseName,
+      String tableName,
+      String tableUuid,
+      OperationType operationType,
+      OperationHistoryStatus status,
+      Instant since,
+      Instant until,
+      int limit);
+}