Skip to content

Commit 78ba0ab

Browse files
mkuchenbeckerclaude
mkuchenbecker
and
claude
authored
feat(optimizer): [2/N] Optimizer REST Service and Controller (#531)
## Optimizer Stack | PR | Content | |---|---| | #527 | Data Model | | #530 | Database Repos | | #531 **(this)** | REST service | | #533 | Analyzer app | | #534 | Scheduler app | | #tbd | Spark BatchedOFD app | | #tbd | Infra, docker-compose, smoke test | ## Summary PR 2 of N in the optimizer stack. Service layer and REST controllers for the optimizer service, plus the `apps/optimizer` shared module providing lightweight entity/repo copies for the analyzer and scheduler apps. ## Changes - [ ] Client-facing API Changes - [x] Internal API Changes - [ ] Bug Fixes - [x] New Features - [ ] Performance Improvements - [ ] Code Style - [ ] Refactoring - [ ] Documentation - [x] Tests **Service layer**: `OptimizerDataService` interface and `OptimizerDataServiceImpl` — CRUD operations, complete-operation lifecycle, stats upsert with history double-write, filtered queries. **Controllers**: `TableOperationsController`, `TableOperationsHistoryController`, `TableStatsController` — REST endpoints per the design doc API spec. **Shared module** (`apps/optimizer`): Lightweight entity and repository copies used by the analyzer and scheduler apps to read optimizer state directly from MySQL. ## Testing Done - [ ] Manually Tested on local docker setup. Please include commands ran, and their output. - [x] Added new tests for the changes made. - [ ] Updated existing tests to reflect the changes made. - [ ] No tests added or updated. Please explain why. If unsure, please feel free to ask for help. - [ ] Some other form of testing like staging or soak time in production. Please explain. H2 integration tests in `OptimizerDataServiceImplTest` (5 tests): - `completeOperation_writesHistoryFromOperationRow` — saves SCHEDULED row, completes it, asserts history DTO fields - `completeOperation_notFound_returnsEmpty` — completes nonexistent ID, asserts empty - `upsertTableStats_createsNewRow` — upserts new table, asserts DTO and repo row - `upsertTableStats_updatesExistingRow` — upserts twice, asserts overwrite with single row - `upsertTableStats_appendsHistoryOnEveryCall` — upserts twice, asserts 2 history rows ``` ./gradlew :services:optimizer:test # BUILD SUCCESSFUL — all 25 tests pass (repo tests from PR 1 + 5 new service tests) ``` # Additional Information - [ ] Breaking Changes - [ ] Deprecations - [x] Large PR broken into smaller PRs, and PR plan linked in the description. --------- Co-authored-by: mkuchenbecker <mkuchenbecker@users.noreply.github.com> Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
1 parent fac85a3 commit 78ba0ab

8 files changed

Lines changed: 885 additions & 0 deletions

File tree

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

Lines changed: 125 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,125 @@
1+
package com.linkedin.openhouse.optimizer.api.controller;
2+
3+
import com.linkedin.openhouse.optimizer.api.spec.OperationStatus;
4+
import com.linkedin.openhouse.optimizer.api.spec.OperationType;
5+
import com.linkedin.openhouse.optimizer.api.spec.TableOperations;
6+
import com.linkedin.openhouse.optimizer.api.spec.TableOperationsHistory;
7+
import com.linkedin.openhouse.optimizer.api.spec.UpdateOperationRequest;
8+
import com.linkedin.openhouse.optimizer.service.OptimizerDataService;
9+
import io.swagger.v3.oas.annotations.responses.ApiResponse;
10+
import io.swagger.v3.oas.annotations.responses.ApiResponses;
11+
import java.util.List;
12+
import java.util.Objects;
13+
import java.util.Optional;
14+
import java.util.stream.Collectors;
15+
import lombok.RequiredArgsConstructor;
16+
import org.springframework.http.HttpStatus;
17+
import org.springframework.http.ResponseEntity;
18+
import org.springframework.util.StringUtils;
19+
import org.springframework.web.bind.annotation.GetMapping;
20+
import org.springframework.web.bind.annotation.PathVariable;
21+
import org.springframework.web.bind.annotation.PostMapping;
22+
import org.springframework.web.bind.annotation.RequestBody;
23+
import org.springframework.web.bind.annotation.RequestMapping;
24+
import org.springframework.web.bind.annotation.RequestParam;
25+
import org.springframework.web.bind.annotation.RestController;
26+
import org.springframework.web.server.ResponseStatusException;
27+
28+
/** REST controller for {@code table_operations}. */
29+
@RestController
30+
@RequestMapping("/v1/optimizer/operations")
31+
@RequiredArgsConstructor
32+
public class TableOperationsController {
33+
34+
private final OptimizerDataService service;
35+
36+
/**
37+
* Report an update to an operation. {@code id} comes from the URL; the body's {@code operationId}
38+
* must match (the controller rejects mismatched requests with 400). The backend looks up the
39+
* operation row, writes a history entry with the operation's table metadata, and returns 201
40+
* Created with the history row, or 404 if the operation does not exist.
41+
*/
42+
@ApiResponses(
43+
value = {
44+
@ApiResponse(responseCode = "201", description = "Operation UPDATE: CREATED"),
45+
@ApiResponse(responseCode = "400", description = "Operation UPDATE: BAD_REQUEST"),
46+
@ApiResponse(responseCode = "404", description = "Operation UPDATE: NOT_FOUND")
47+
})
48+
@PostMapping("/{id}/update")
49+
public ResponseEntity<TableOperationsHistory> updateOperation(
50+
@PathVariable String id, @RequestBody UpdateOperationRequest request) {
51+
if (!StringUtils.hasText(request.getOperationId())) {
52+
throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "operationId is required");
53+
}
54+
if (!Objects.equals(id, request.getOperationId())) {
55+
throw new ResponseStatusException(
56+
HttpStatus.BAD_REQUEST,
57+
String.format(
58+
"operationId in body (%s) does not match path id (%s)",
59+
request.getOperationId(), id));
60+
}
61+
if (request.getStatus() == null) {
62+
throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "status is required");
63+
}
64+
return service
65+
.updateOperation(id, request.getStatus().toModel())
66+
.map(
67+
history ->
68+
ResponseEntity.status(HttpStatus.CREATED)
69+
.body(TableOperationsHistory.fromModel(history)))
70+
.orElseThrow(
71+
() ->
72+
new ResponseStatusException(
73+
HttpStatus.NOT_FOUND, String.format("no operation with id %s", id)));
74+
}
75+
76+
/** Fetch a single operation row by its ID, regardless of status. Returns 404 if not found. */
77+
@ApiResponses(
78+
value = {
79+
@ApiResponse(responseCode = "200", description = "Operation GET: OK"),
80+
@ApiResponse(responseCode = "404", description = "Operation GET: NOT_FOUND")
81+
})
82+
@GetMapping("/{id}")
83+
public ResponseEntity<TableOperations> getTableOperation(@PathVariable String id) {
84+
return service
85+
.getTableOperation(id)
86+
.map(TableOperations::fromModel)
87+
.map(ResponseEntity::ok)
88+
.orElseThrow(
89+
() ->
90+
new ResponseStatusException(
91+
HttpStatus.NOT_FOUND, String.format("no operation with id %s", id)));
92+
}
93+
94+
/**
95+
* List operations matching the given filters, capped at {@code limit} rows. Every filter is
96+
* optional; {@code limit} is required so callers always state how much they want back.
97+
*/
98+
@ApiResponses(
99+
value = {
100+
@ApiResponse(responseCode = "200", description = "Operation SEARCH: OK"),
101+
@ApiResponse(responseCode = "400", description = "Operation SEARCH: BAD_REQUEST")
102+
})
103+
@GetMapping
104+
public ResponseEntity<List<TableOperations>> listTableOperations(
105+
@RequestParam(required = false) OperationType operationType,
106+
@RequestParam(required = false) OperationStatus status,
107+
@RequestParam(required = false) String databaseName,
108+
@RequestParam(required = false) String tableName,
109+
@RequestParam(required = false) String tableUuid,
110+
@RequestParam int limit) {
111+
List<TableOperations> result =
112+
service
113+
.listTableOperations(
114+
Optional.ofNullable(operationType).map(OperationType::toModel),
115+
Optional.ofNullable(status).map(OperationStatus::toModel),
116+
Optional.ofNullable(databaseName),
117+
Optional.ofNullable(tableName),
118+
Optional.ofNullable(tableUuid),
119+
limit)
120+
.stream()
121+
.map(TableOperations::fromModel)
122+
.collect(Collectors.toList());
123+
return ResponseEntity.ok(result);
124+
}
125+
}

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

Lines changed: 58 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,58 @@
1+
package com.linkedin.openhouse.optimizer.api.controller;
2+
3+
import com.linkedin.openhouse.optimizer.api.spec.TableOperationsHistory;
4+
import com.linkedin.openhouse.optimizer.service.OptimizerDataService;
5+
import io.swagger.v3.oas.annotations.responses.ApiResponse;
6+
import io.swagger.v3.oas.annotations.responses.ApiResponses;
7+
import java.util.List;
8+
import java.util.stream.Collectors;
9+
import lombok.RequiredArgsConstructor;
10+
import org.springframework.http.HttpStatus;
11+
import org.springframework.http.ResponseEntity;
12+
import org.springframework.web.bind.annotation.GetMapping;
13+
import org.springframework.web.bind.annotation.PathVariable;
14+
import org.springframework.web.bind.annotation.PostMapping;
15+
import org.springframework.web.bind.annotation.RequestBody;
16+
import org.springframework.web.bind.annotation.RequestMapping;
17+
import org.springframework.web.bind.annotation.RequestParam;
18+
import org.springframework.web.bind.annotation.RestController;
19+
20+
/** REST controller for {@code table_operations_history}. */
21+
@RestController
22+
@RequestMapping("/v1/optimizer/operations-history")
23+
@RequiredArgsConstructor
24+
public class TableOperationsHistoryController {
25+
26+
private final OptimizerDataService service;
27+
28+
/** Append a completed-job result. Called by the SparkJob after each run (success or failure). */
29+
@ApiResponses(
30+
value = {
31+
@ApiResponse(responseCode = "201", description = "OperationsHistory CREATE: CREATED")
32+
})
33+
@PostMapping
34+
public ResponseEntity<TableOperationsHistory> appendHistory(
35+
@RequestBody TableOperationsHistory dto) {
36+
return ResponseEntity.status(HttpStatus.CREATED)
37+
.body(TableOperationsHistory.fromModel(service.appendHistory(dto.toModel())));
38+
}
39+
40+
/**
41+
* Return the most recent history for a table, newest first, capped at {@code limit} rows. {@code
42+
* limit} is required.
43+
*/
44+
@ApiResponses(
45+
value = {
46+
@ApiResponse(responseCode = "200", description = "OperationsHistory GET: OK"),
47+
@ApiResponse(responseCode = "400", description = "OperationsHistory GET: BAD_REQUEST")
48+
})
49+
@GetMapping("/{tableUuid}")
50+
public ResponseEntity<List<TableOperationsHistory>> getHistory(
51+
@PathVariable String tableUuid, @RequestParam int limit) {
52+
List<TableOperationsHistory> result =
53+
service.getHistory(tableUuid, limit).stream()
54+
.map(TableOperationsHistory::fromModel)
55+
.collect(Collectors.toList());
56+
return ResponseEntity.ok(result);
57+
}
58+
}

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

Lines changed: 111 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,111 @@
1+
package com.linkedin.openhouse.optimizer.api.controller;
2+
3+
import com.linkedin.openhouse.optimizer.api.spec.TableStats;
4+
import com.linkedin.openhouse.optimizer.api.spec.TableStatsHistory;
5+
import com.linkedin.openhouse.optimizer.api.spec.UpsertTableStatsRequest;
6+
import com.linkedin.openhouse.optimizer.service.OptimizerDataService;
7+
import io.swagger.v3.oas.annotations.responses.ApiResponse;
8+
import io.swagger.v3.oas.annotations.responses.ApiResponses;
9+
import java.time.Instant;
10+
import java.util.List;
11+
import java.util.Optional;
12+
import java.util.stream.Collectors;
13+
import lombok.RequiredArgsConstructor;
14+
import org.springframework.http.HttpStatus;
15+
import org.springframework.http.ResponseEntity;
16+
import org.springframework.web.bind.annotation.GetMapping;
17+
import org.springframework.web.bind.annotation.PathVariable;
18+
import org.springframework.web.bind.annotation.PutMapping;
19+
import org.springframework.web.bind.annotation.RequestBody;
20+
import org.springframework.web.bind.annotation.RequestMapping;
21+
import org.springframework.web.bind.annotation.RequestParam;
22+
import org.springframework.web.bind.annotation.RestController;
23+
import org.springframework.web.server.ResponseStatusException;
24+
25+
/** REST controller for managing per-table stats in the optimizer DB. */
26+
@RestController
27+
@RequestMapping("/v1/optimizer/stats")
28+
@RequiredArgsConstructor
29+
public class TableStatsController {
30+
31+
private final OptimizerDataService service;
32+
33+
/**
34+
* Create or overwrite the stats row for {@code tableUuid}. Called by the Tables Service on every
35+
* Iceberg commit. Idempotent.
36+
*/
37+
@ApiResponses(value = {@ApiResponse(responseCode = "200", description = "Stats PUT: OK")})
38+
@PutMapping("/{tableUuid}")
39+
public ResponseEntity<TableStats> upsertTableStats(
40+
@PathVariable String tableUuid, @RequestBody UpsertTableStatsRequest request) {
41+
return ResponseEntity.ok(
42+
TableStats.fromModel(service.upsertTableStats(request.toModel(tableUuid))));
43+
}
44+
45+
/** Fetch the stats row for {@code tableUuid}. Returns 404 if no stats have been written yet. */
46+
@ApiResponses(
47+
value = {
48+
@ApiResponse(responseCode = "200", description = "Stats GET: OK"),
49+
@ApiResponse(responseCode = "404", description = "Stats GET: NOT_FOUND")
50+
})
51+
@GetMapping("/{tableUuid}")
52+
public ResponseEntity<TableStats> getTableStats(@PathVariable String tableUuid) {
53+
return service
54+
.getTableStats(tableUuid)
55+
.map(TableStats::fromModel)
56+
.map(ResponseEntity::ok)
57+
.orElseThrow(
58+
() ->
59+
new ResponseStatusException(
60+
HttpStatus.NOT_FOUND, String.format("no stats for tableUuid %s", tableUuid)));
61+
}
62+
63+
/**
64+
* List stats rows matching the given filters, capped at {@code limit} rows. Every filter is
65+
* optional; {@code limit} is required so callers always state how much they want back.
66+
*/
67+
@ApiResponses(
68+
value = {
69+
@ApiResponse(responseCode = "200", description = "Stats SEARCH: OK"),
70+
@ApiResponse(responseCode = "400", description = "Stats SEARCH: BAD_REQUEST")
71+
})
72+
@GetMapping
73+
public ResponseEntity<List<TableStats>> listTableStats(
74+
@RequestParam(required = false) String databaseName,
75+
@RequestParam(required = false) String tableName,
76+
@RequestParam(required = false) String tableUuid,
77+
@RequestParam int limit) {
78+
List<TableStats> result =
79+
service
80+
.listTableStats(
81+
Optional.ofNullable(databaseName),
82+
Optional.ofNullable(tableName),
83+
Optional.ofNullable(tableUuid),
84+
limit)
85+
.stream()
86+
.map(TableStats::fromModel)
87+
.collect(Collectors.toList());
88+
return ResponseEntity.ok(result);
89+
}
90+
91+
/**
92+
* Return per-commit stats history for {@code tableUuid}, newest first, capped at {@code limit}
93+
* rows. Optional {@code since} filter (inclusive). {@code limit} is required.
94+
*/
95+
@ApiResponses(
96+
value = {
97+
@ApiResponse(responseCode = "200", description = "StatsHistory GET: OK"),
98+
@ApiResponse(responseCode = "400", description = "StatsHistory GET: BAD_REQUEST")
99+
})
100+
@GetMapping("/{tableUuid}/history")
101+
public ResponseEntity<List<TableStatsHistory>> getStatsHistory(
102+
@PathVariable String tableUuid,
103+
@RequestParam(required = false) Instant since,
104+
@RequestParam int limit) {
105+
List<TableStatsHistory> result =
106+
service.getStatsHistory(tableUuid, Optional.ofNullable(since), limit).stream()
107+
.map(TableStatsHistory::fromModel)
108+
.collect(Collectors.toList());
109+
return ResponseEntity.ok(result);
110+
}
111+
}

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

Lines changed: 94 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,94 @@
1+
package com.linkedin.openhouse.optimizer.service;
2+
3+
import com.linkedin.openhouse.optimizer.model.HistoryStatusDto;
4+
import com.linkedin.openhouse.optimizer.model.OperationStatusDto;
5+
import com.linkedin.openhouse.optimizer.model.OperationTypeDto;
6+
import com.linkedin.openhouse.optimizer.model.TableOperationDto;
7+
import com.linkedin.openhouse.optimizer.model.TableOperationsHistoryDto;
8+
import com.linkedin.openhouse.optimizer.model.TableStatsDto;
9+
import com.linkedin.openhouse.optimizer.model.TableStatsHistoryDto;
10+
import java.time.Instant;
11+
import java.util.List;
12+
import java.util.Optional;
13+
14+
/**
15+
* Service interface for optimizer data operations.
16+
*
17+
* <p>The service is the boundary between the wire-API surface and the database. Inputs and outputs
18+
* are <em>internal-model</em> types only — callers (controllers, future CLI, in-process consumers)
19+
* convert at their own edge. No api/-package types appear here.
20+
*/
21+
public interface OptimizerDataService {
22+
23+
// --- TableOperations ---
24+
25+
/**
26+
* List operations matching the given filters, capped at {@code limit} rows. Every filter
27+
* parameter is optional — pass {@link Optional#empty()} to skip that filter.
28+
*/
29+
List<TableOperationDto> listTableOperations(
30+
Optional<OperationTypeDto> operationType,
31+
Optional<OperationStatusDto> status,
32+
Optional<String> databaseName,
33+
Optional<String> tableName,
34+
Optional<String> tableUuid,
35+
int limit);
36+
37+
/**
38+
* Update an operation by writing a history entry. Looks up the operation row by {@code
39+
* operationId}, copies its table metadata into a new history row with the supplied terminal
40+
* {@code status}, and saves it. Returns the history record, or empty if the operation does not
41+
* exist.
42+
*/
43+
Optional<TableOperationsHistoryDto> updateOperation(String operationId, HistoryStatusDto status);
44+
45+
/**
46+
* Return the operation row for {@code id} regardless of status, or empty if it does not exist.
47+
* Used to poll a specific operation (e.g. waiting for SUCCESS after a Spark job completes).
48+
*/
49+
Optional<TableOperationDto> getTableOperation(String id);
50+
51+
// --- TableStatsDto ---
52+
53+
/**
54+
* Create or update the stats row for {@code stats.getTableUuid()}. Fully idempotent: the same
55+
* call overwrites the previous snapshot with the latest commit values. The service stamps {@link
56+
* TableStatsDto#getUpdatedAt()} server-side and returns the resulting {@link TableStatsDto}.
57+
*/
58+
TableStatsDto upsertTableStats(TableStatsDto stats);
59+
60+
/** Return the stats row for {@code tableUuid}, or empty if none exists. */
61+
Optional<TableStatsDto> getTableStats(String tableUuid);
62+
63+
/**
64+
* List stats rows matching the given filters, capped at {@code limit} rows. Every filter
65+
* parameter is optional — pass {@link Optional#empty()} to skip that filter.
66+
*/
67+
List<TableStatsDto> listTableStats(
68+
Optional<String> databaseName,
69+
Optional<String> tableName,
70+
Optional<String> tableUuid,
71+
int limit);
72+
73+
/**
74+
* Return per-commit stats history for {@code tableUuid}, newest first.
75+
*
76+
* @param tableUuid the stable table UUID
77+
* @param since if present, only return rows recorded at or after this instant
78+
* @param limit maximum number of rows to return
79+
*/
80+
List<TableStatsHistoryDto> getStatsHistory(String tableUuid, Optional<Instant> since, int limit);
81+
82+
// --- TableOperationsHistoryDto ---
83+
84+
/** Append a completed-job result record. */
85+
TableOperationsHistoryDto appendHistory(TableOperationsHistoryDto history);
86+
87+
/**
88+
* Return the most recent history rows for a table UUID, newest first.
89+
*
90+
* @param tableUuid the stable table UUID
91+
* @param limit maximum number of rows to return
92+
*/
93+
List<TableOperationsHistoryDto> getHistory(String tableUuid, int limit);
94+
}

0 commit comments

Comments
 (0)