feat(optimizer): [2/N] Optimizer REST Service and Controller

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

@@ -0,0 +1,125 @@
+package com.linkedin.openhouse.optimizer.api.controller;
+
+import com.linkedin.openhouse.optimizer.api.spec.OperationStatus;
+import com.linkedin.openhouse.optimizer.api.spec.OperationType;
+import com.linkedin.openhouse.optimizer.api.spec.TableOperations;
+import com.linkedin.openhouse.optimizer.api.spec.TableOperationsHistory;
+import com.linkedin.openhouse.optimizer.api.spec.UpdateOperationRequest;
+import com.linkedin.openhouse.optimizer.service.OptimizerDataService;
+import io.swagger.v3.oas.annotations.responses.ApiResponse;
+import io.swagger.v3.oas.annotations.responses.ApiResponses;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import lombok.RequiredArgsConstructor;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.util.StringUtils;
+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;
+import org.springframework.web.server.ResponseStatusException;
+
+/** REST controller for {@code table_operations}. */
+@RestController
+@RequestMapping("/v1/optimizer/operations")
+@RequiredArgsConstructor
+public class TableOperationsController {
+
+  private final OptimizerDataService service;
+
+  /**
+   * Report an update to an operation. {@code id} comes from the URL; the body's {@code operationId}
+   * must match (the controller rejects mismatched requests with 400). The backend looks up the
+   * operation row, writes a history entry with the operation's table metadata, and returns 201
+   * Created with the history row, or 404 if the operation does not exist.
+   */
+  @ApiResponses(
+      value = {
+        @ApiResponse(responseCode = "201", description = "Operation UPDATE: CREATED"),
+        @ApiResponse(responseCode = "400", description = "Operation UPDATE: BAD_REQUEST"),
+        @ApiResponse(responseCode = "404", description = "Operation UPDATE: NOT_FOUND")
+      })
+  @PostMapping("/{id}/update")
+  public ResponseEntity<TableOperationsHistory> updateOperation(
+      @PathVariable String id, @RequestBody UpdateOperationRequest request) {
+    if (!StringUtils.hasText(request.getOperationId())) {
+      throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "operationId is required");
+    }
+    if (!Objects.equals(id, request.getOperationId())) {
+      throw new ResponseStatusException(
+          HttpStatus.BAD_REQUEST,
+          String.format(
+              "operationId in body (%s) does not match path id (%s)",
+              request.getOperationId(), id));
+    }
+    if (request.getStatus() == null) {
+      throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "status is required");
+    }
+    return service
+        .updateOperation(id, request.getStatus().toModel())
+        .map(
+            history ->
+                ResponseEntity.status(HttpStatus.CREATED)
+                    .body(TableOperationsHistory.fromModel(history)))
+        .orElseThrow(
+            () ->
+                new ResponseStatusException(
+                    HttpStatus.NOT_FOUND, String.format("no operation with id %s", id)));
+  }
+
+  /** Fetch a single operation row by its ID, regardless of status. Returns 404 if not found. */
+  @ApiResponses(
+      value = {
+        @ApiResponse(responseCode = "200", description = "Operation GET: OK"),
+        @ApiResponse(responseCode = "404", description = "Operation GET: NOT_FOUND")
+      })
+  @GetMapping("/{id}")
+  public ResponseEntity<TableOperations> getTableOperation(@PathVariable String id) {
+    return service
+        .getTableOperation(id)
+        .map(TableOperations::fromModel)
+        .map(ResponseEntity::ok)
+        .orElseThrow(
+            () ->
+                new ResponseStatusException(
+                    HttpStatus.NOT_FOUND, String.format("no operation with id %s", id)));
+  }
+
+  /**
+   * List operations matching the given filters, capped at {@code limit} rows. Every filter is
+   * optional; {@code limit} is required so callers always state how much they want back.
+   */
+  @ApiResponses(
+      value = {
+        @ApiResponse(responseCode = "200", description = "Operation SEARCH: OK"),
+        @ApiResponse(responseCode = "400", description = "Operation SEARCH: BAD_REQUEST")
+      })
+  @GetMapping
+  public ResponseEntity<List<TableOperations>> 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,
+      @RequestParam int limit) {
+    List<TableOperations> result =
+        service
+            .listTableOperations(
+                Optional.ofNullable(operationType).map(OperationType::toModel),
+                Optional.ofNullable(status).map(OperationStatus::toModel),
+                Optional.ofNullable(databaseName),
+                Optional.ofNullable(tableName),
+                Optional.ofNullable(tableUuid),
+                limit)
+            .stream()
+            .map(TableOperations::fromModel)
+            .collect(Collectors.toList());
+    return ResponseEntity.ok(result);
+  }
+}

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

@@ -0,0 +1,58 @@
+package com.linkedin.openhouse.optimizer.api.controller;
+
+import com.linkedin.openhouse.optimizer.api.spec.TableOperationsHistory;
+import com.linkedin.openhouse.optimizer.service.OptimizerDataService;
+import io.swagger.v3.oas.annotations.responses.ApiResponse;
+import io.swagger.v3.oas.annotations.responses.ApiResponses;
+import java.util.List;
+import java.util.stream.Collectors;
+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/optimizer/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). */
+  @ApiResponses(
+      value = {
+        @ApiResponse(responseCode = "201", description = "OperationsHistory CREATE: CREATED")
+      })
+  @PostMapping
+  public ResponseEntity<TableOperationsHistory> appendHistory(
+      @RequestBody TableOperationsHistory dto) {
+    return ResponseEntity.status(HttpStatus.CREATED)
+        .body(TableOperationsHistory.fromModel(service.appendHistory(dto.toModel())));
+  }
+
+  /**
+   * Return the most recent history for a table, newest first, capped at {@code limit} rows. {@code
+   * limit} is required.
+   */
+  @ApiResponses(
+      value = {
+        @ApiResponse(responseCode = "200", description = "OperationsHistory GET: OK"),
+        @ApiResponse(responseCode = "400", description = "OperationsHistory GET: BAD_REQUEST")
+      })
+  @GetMapping("/{tableUuid}")
+  public ResponseEntity<List<TableOperationsHistory>> getHistory(
+      @PathVariable String tableUuid, @RequestParam int limit) {
+    List<TableOperationsHistory> result =
+        service.getHistory(tableUuid, limit).stream()
+            .map(TableOperationsHistory::fromModel)
+            .collect(Collectors.toList());
+    return ResponseEntity.ok(result);
+  }
+}

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

@@ -0,0 +1,111 @@
+package com.linkedin.openhouse.optimizer.api.controller;
+
+import com.linkedin.openhouse.optimizer.api.spec.TableStats;
+import com.linkedin.openhouse.optimizer.api.spec.TableStatsHistory;
+import com.linkedin.openhouse.optimizer.api.spec.UpsertTableStatsRequest;
+import com.linkedin.openhouse.optimizer.service.OptimizerDataService;
+import io.swagger.v3.oas.annotations.responses.ApiResponse;
+import io.swagger.v3.oas.annotations.responses.ApiResponses;
+import java.time.Instant;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+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.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;
+import org.springframework.web.server.ResponseStatusException;
+
+/** REST controller for managing per-table stats in the optimizer DB. */
+@RestController
+@RequestMapping("/v1/optimizer/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.
+   */
+  @ApiResponses(value = {@ApiResponse(responseCode = "200", description = "Stats PUT: OK")})
+  @PutMapping("/{tableUuid}")
+  public ResponseEntity<TableStats> upsertTableStats(
+      @PathVariable String tableUuid, @RequestBody UpsertTableStatsRequest request) {
+    return ResponseEntity.ok(
+        TableStats.fromModel(service.upsertTableStats(request.toModel(tableUuid))));
+  }
+
+  /** Fetch the stats row for {@code tableUuid}. Returns 404 if no stats have been written yet. */
+  @ApiResponses(
+      value = {
+        @ApiResponse(responseCode = "200", description = "Stats GET: OK"),
+        @ApiResponse(responseCode = "404", description = "Stats GET: NOT_FOUND")
+      })
+  @GetMapping("/{tableUuid}")
+  public ResponseEntity<TableStats> getTableStats(@PathVariable String tableUuid) {
+    return service
+        .getTableStats(tableUuid)
+        .map(TableStats::fromModel)
+        .map(ResponseEntity::ok)
+        .orElseThrow(
+            () ->
+                new ResponseStatusException(
+                    HttpStatus.NOT_FOUND, String.format("no stats for tableUuid %s", tableUuid)));
+  }
+
+  /**
+   * List stats rows matching the given filters, capped at {@code limit} rows. Every filter is
+   * optional; {@code limit} is required so callers always state how much they want back.
+   */
+  @ApiResponses(
+      value = {
+        @ApiResponse(responseCode = "200", description = "Stats SEARCH: OK"),
+        @ApiResponse(responseCode = "400", description = "Stats SEARCH: BAD_REQUEST")
+      })
+  @GetMapping
+  public ResponseEntity<List<TableStats>> listTableStats(
+      @RequestParam(required = false) String databaseName,
+      @RequestParam(required = false) String tableName,
+      @RequestParam(required = false) String tableUuid,
+      @RequestParam int limit) {
+    List<TableStats> result =
+        service
+            .listTableStats(
+                Optional.ofNullable(databaseName),
+                Optional.ofNullable(tableName),
+                Optional.ofNullable(tableUuid),
+                limit)
+            .stream()
+            .map(TableStats::fromModel)
+            .collect(Collectors.toList());
+    return ResponseEntity.ok(result);
+  }
+
+  /**
+   * Return per-commit stats history for {@code tableUuid}, newest first, capped at {@code limit}
+   * rows. Optional {@code since} filter (inclusive). {@code limit} is required.
+   */
+  @ApiResponses(
+      value = {
+        @ApiResponse(responseCode = "200", description = "StatsHistory GET: OK"),
+        @ApiResponse(responseCode = "400", description = "StatsHistory GET: BAD_REQUEST")
+      })
+  @GetMapping("/{tableUuid}/history")
+  public ResponseEntity<List<TableStatsHistory>> getStatsHistory(
+      @PathVariable String tableUuid,
+      @RequestParam(required = false) Instant since,
+      @RequestParam int limit) {
+    List<TableStatsHistory> result =
+        service.getStatsHistory(tableUuid, Optional.ofNullable(since), limit).stream()
+            .map(TableStatsHistory::fromModel)
+            .collect(Collectors.toList());
+    return ResponseEntity.ok(result);
+  }
+}

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

@@ -0,0 +1,94 @@
+package com.linkedin.openhouse.optimizer.service;
+
+import com.linkedin.openhouse.optimizer.model.HistoryStatusDto;
+import com.linkedin.openhouse.optimizer.model.OperationStatusDto;
+import com.linkedin.openhouse.optimizer.model.OperationTypeDto;
+import com.linkedin.openhouse.optimizer.model.TableOperationDto;
+import com.linkedin.openhouse.optimizer.model.TableOperationsHistoryDto;
+import com.linkedin.openhouse.optimizer.model.TableStatsDto;
+import com.linkedin.openhouse.optimizer.model.TableStatsHistoryDto;
+import java.time.Instant;
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * Service interface for optimizer data operations.
+ *
+ * <p>The service is the boundary between the wire-API surface and the database. Inputs and outputs
+ * are <em>internal-model</em> types only — callers (controllers, future CLI, in-process consumers)
+ * convert at their own edge. No api/-package types appear here.
+ */
+public interface OptimizerDataService {
+
+  // --- TableOperations ---
+
+  /**
+   * List operations matching the given filters, capped at {@code limit} rows. Every filter
+   * parameter is optional — pass {@link Optional#empty()} to skip that filter.
+   */
+  List<TableOperationDto> listTableOperations(
+      Optional<OperationTypeDto> operationType,
+      Optional<OperationStatusDto> status,
+      Optional<String> databaseName,
+      Optional<String> tableName,
+      Optional<String> tableUuid,
+      int limit);
+
+  /**
+   * Update an operation by writing a history entry. Looks up the operation row by {@code
+   * operationId}, copies its table metadata into a new history row with the supplied terminal
+   * {@code status}, and saves it. Returns the history record, or empty if the operation does not
+   * exist.
+   */
+  Optional<TableOperationsHistoryDto> updateOperation(String operationId, HistoryStatusDto status);
+
+  /**
+   * 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<TableOperationDto> getTableOperation(String id);
+
+  // --- TableStatsDto ---
+
+  /**
+   * Create or update the stats row for {@code stats.getTableUuid()}. Fully idempotent: the same
+   * call overwrites the previous snapshot with the latest commit values. The service stamps {@link
+   * TableStatsDto#getUpdatedAt()} server-side and returns the resulting {@link TableStatsDto}.
+   */
+  TableStatsDto upsertTableStats(TableStatsDto stats);
+
+  /** Return the stats row for {@code tableUuid}, or empty if none exists. */
+  Optional<TableStatsDto> getTableStats(String tableUuid);
+
+  /**
+   * List stats rows matching the given filters, capped at {@code limit} rows. Every filter
+   * parameter is optional — pass {@link Optional#empty()} to skip that filter.
+   */
+  List<TableStatsDto> listTableStats(
+      Optional<String> databaseName,
+      Optional<String> tableName,
+      Optional<String> tableUuid,
+      int limit);
+
+  /**
+   * Return per-commit stats history for {@code tableUuid}, newest first.
+   *
+   * @param tableUuid the stable table UUID
+   * @param since if present, only return rows recorded at or after this instant
+   * @param limit maximum number of rows to return
+   */
+  List<TableStatsHistoryDto> getStatsHistory(String tableUuid, Optional<Instant> since, int limit);
+
+  // --- TableOperationsHistoryDto ---
+
+  /** Append a completed-job result record. */
+  TableOperationsHistoryDto appendHistory(TableOperationsHistoryDto history);
+
+  /**
+   * 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);
+}