Yicong-Huang commented on code in PR #8552: URL: https://github.com/apache/texera/pull/8552#discussion_r4048986378
########## frontend/src/app/workspace/service/heatmap/runtime-statistics-mapper.ts: ########## @@ -0,0 +1,80 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { OperatorRuntimeStatus, OperatorState } from "../../types/execute-workflow.interface"; +import { WorkflowRuntimeStatistics } from "../../../dashboard/type/workflow-runtime-statistics"; + +/** + * Maps a persisted per-operator status code back to an OperatorState. + * + * The engine persists the codes written by Utils.maptoStatusCode: 0 = + * Uninitialized/Ready, 1 = Running, 2 = Paused, 3 = Completed. Codes 4 + * (Failed), 5 (Killed) and -1 (other) have no OperatorState member, so they + * fall back to the neutral Uninitialized rather than a wrong state. + */ +export function operatorStateFromStatusCode(code: number): OperatorState { + switch (code) { + case 0: + return OperatorState.Uninitialized; + case 1: + return OperatorState.Running; + case 2: + return OperatorState.Paused; + case 3: + return OperatorState.Completed; + default: + return OperatorState.Uninitialized; + } +} + +/** + * Reduces a persisted runtime-statistics time series to the latest snapshot + * per operator (by timestamp; the later row wins a tie, matching write + * order), mapped to the wire shape WorkflowStatusService ingests. + * + * Port-level metrics are not persisted historically, so the port maps are + * empty — restored port labels read 0, like a resetStatus snapshot. + */ +export function toOperatorRuntimeStatusMap(rows: WorkflowRuntimeStatistics[]): Record<string, OperatorRuntimeStatus> { + const latestByOperator: Record<string, WorkflowRuntimeStatistics> = {}; + for (const row of rows) { + const seen = latestByOperator[row.operatorId]; + if (seen === undefined || row.timestamp >= seen.timestamp) { + latestByOperator[row.operatorId] = row; + } + } + + const result: Record<string, OperatorRuntimeStatus> = {}; + for (const [operatorId, row] of Object.entries(latestByOperator)) { + result[operatorId] = { + operatorState: operatorStateFromStatusCode(row.status), + aggregatedInputRowCount: row.inputTupleCount, + aggregatedInputSize: row.inputTupleSize, + inputPortMetrics: {}, Review Comment: **Must fix:** The persisted row has no per-port column (`ExecutionStatsService.scala:270-273` sums them away), so these maps are empty. `JointUIService` reads a missing port as `?? 0` (`joint-ui.service.ts:400`, `:410`) and writes `"0"` over the port display names set at `:336`. Reload after a run and every label reads `0` beside correct heat-map colours. It sticks: this snapshot becomes `getCurrentStatistics()` and is reapplied at `workflow-editor.component.ts:420`. Leaving the port maps undefined would keep the labels intact. ########## frontend/src/app/workspace/service/heatmap/heatmap-stats-restore.service.ts: ########## @@ -0,0 +1,105 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { Injectable } from "@angular/core"; +import { EMPTY, Observable, of } from "rxjs"; +import { catchError, map, switchMap } from "rxjs/operators"; +import { WorkflowExecutionsService } from "../../../dashboard/service/user/workflow-executions/workflow-executions.service"; +import { WorkflowExecutionsEntry } from "../../../dashboard/type/workflow-executions-entry"; +import { WorkflowActionService } from "../workflow-graph/model/workflow-action.service"; +import { WorkflowStatusService } from "../workflow-status/workflow-status.service"; +import { ExecuteWorkflowService } from "../execute-workflow/execute-workflow.service"; +import { isNotInExecution } from "../../types/execute-workflow.interface"; +import { loadPersistedHeatmapView } from "./heatmap-overlay-persistence"; +import { toOperatorRuntimeStatusMap } from "./runtime-statistics-mapper"; + +/** Persisted execution status code for a completed run (EXECUTION_STATUS_CODE). */ +const EXECUTION_COMPLETED_CODE = 3; + +/** + * Restores the last execution's per-operator statistics after a page refresh, + * so the performance heat-map can render a finished run without re-executing. + * + * All gating lives here rather than in the caller: the persisted-overlay check + * is a synchronous localStorage read, so a user who never enabled the overlay + * incurs no fetch at all. + */ +@Injectable({ + providedIn: "root", +}) +export class HeatmapStatsRestoreService { + constructor( + private workflowExecutionsService: WorkflowExecutionsService, + private workflowActionService: WorkflowActionService, + private workflowStatusService: WorkflowStatusService, + private executeWorkflowService: ExecuteWorkflowService + ) {} + + /** + * Fetches the latest run's statistics and feeds them into + * WorkflowStatusService. Cold: nothing happens until subscribed. Skips + * silently (including on HTTP errors — restoring is best-effort) when: + * - the overlay is not persisted on, + * - the workflow has never been saved (no wid), + * - an execution is in progress (the live stream wins), + * - the workflow has no executions or the run left no statistics. + */ + public restoreLatestRunStatistics(): Observable<void> { + if (loadPersistedHeatmapView() === null) { + return EMPTY; + } + const wid = this.workflowActionService.getWorkflowMetadata()?.wid; + if (wid === undefined) { + return EMPTY; + } + if (!isNotInExecution(this.executeWorkflowService.getExecutionState().state)) { Review Comment: **Must fix:** These gates run eagerly, two HTTP round trips before the result is used, despite the docstring at `:56`. `currentState` starts `Uninitialized` (`execute-workflow.service.ts:82`) and `isNotInExecution` is true for it (`execute-workflow.interface.ts:152`), so this passes while the page is still connecting. A reconnecting session does receive a live snapshot (`StateStore.scala:34-45`), so a mid-run refresh could overwrite it with a different execution's data. Does the socket land inside that window in practice? Re-checking before `setExternalStatus` would close it either way. ########## frontend/src/app/workspace/service/heatmap/heatmap-stats-restore.service.ts: ########## @@ -0,0 +1,105 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { Injectable } from "@angular/core"; +import { EMPTY, Observable, of } from "rxjs"; +import { catchError, map, switchMap } from "rxjs/operators"; +import { WorkflowExecutionsService } from "../../../dashboard/service/user/workflow-executions/workflow-executions.service"; +import { WorkflowExecutionsEntry } from "../../../dashboard/type/workflow-executions-entry"; +import { WorkflowActionService } from "../workflow-graph/model/workflow-action.service"; +import { WorkflowStatusService } from "../workflow-status/workflow-status.service"; +import { ExecuteWorkflowService } from "../execute-workflow/execute-workflow.service"; +import { isNotInExecution } from "../../types/execute-workflow.interface"; +import { loadPersistedHeatmapView } from "./heatmap-overlay-persistence"; +import { toOperatorRuntimeStatusMap } from "./runtime-statistics-mapper"; + +/** Persisted execution status code for a completed run (EXECUTION_STATUS_CODE). */ +const EXECUTION_COMPLETED_CODE = 3; Review Comment: **Advisory:** `EXECUTION_STATUS_CODE` at `workflow-executions-entry.ts:37` already holds this, in the module line 24 already imports from, and the comment above even names it. Nothing misbehaves today. But the codes are owned Scala-side, so a renumber there would leave this literal compiling while silently selecting a different execution. ########## frontend/src/app/workspace/service/heatmap/heatmap-overlay-persistence.ts: ########## @@ -0,0 +1,51 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { localGetObject, localSetObject } from "../../../common/util/storage"; +import { HeatmapView } from "./heatmap-scoring"; + +/** localStorage key for the persisted heat-map overlay state. */ +export const HEATMAP_OVERLAY_STORAGE_KEY = "heatmapOverlay"; Review Comment: **Polish:** This is the only camelCase localStorage key among the frontend's 16, which are otherwise kebab-case (`mini-map`, `left-panel-order`). Since it is persisted in users' browsers, renaming later means orphaning existing values or writing a migration. ```suggestion export const HEATMAP_OVERLAY_STORAGE_KEY = "heatmap-overlay"; ``` ########## frontend/src/app/workspace/service/heatmap/heatmap-stats-restore.service.ts: ########## @@ -0,0 +1,105 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { Injectable } from "@angular/core"; +import { EMPTY, Observable, of } from "rxjs"; +import { catchError, map, switchMap } from "rxjs/operators"; +import { WorkflowExecutionsService } from "../../../dashboard/service/user/workflow-executions/workflow-executions.service"; +import { WorkflowExecutionsEntry } from "../../../dashboard/type/workflow-executions-entry"; +import { WorkflowActionService } from "../workflow-graph/model/workflow-action.service"; +import { WorkflowStatusService } from "../workflow-status/workflow-status.service"; +import { ExecuteWorkflowService } from "../execute-workflow/execute-workflow.service"; +import { isNotInExecution } from "../../types/execute-workflow.interface"; +import { loadPersistedHeatmapView } from "./heatmap-overlay-persistence"; +import { toOperatorRuntimeStatusMap } from "./runtime-statistics-mapper"; + +/** Persisted execution status code for a completed run (EXECUTION_STATUS_CODE). */ +const EXECUTION_COMPLETED_CODE = 3; + +/** + * Restores the last execution's per-operator statistics after a page refresh, + * so the performance heat-map can render a finished run without re-executing. + * + * All gating lives here rather than in the caller: the persisted-overlay check + * is a synchronous localStorage read, so a user who never enabled the overlay + * incurs no fetch at all. + */ +@Injectable({ + providedIn: "root", +}) +export class HeatmapStatsRestoreService { + constructor( + private workflowExecutionsService: WorkflowExecutionsService, + private workflowActionService: WorkflowActionService, + private workflowStatusService: WorkflowStatusService, + private executeWorkflowService: ExecuteWorkflowService + ) {} + + /** + * Fetches the latest run's statistics and feeds them into + * WorkflowStatusService. Cold: nothing happens until subscribed. Skips + * silently (including on HTTP errors — restoring is best-effort) when: + * - the overlay is not persisted on, + * - the workflow has never been saved (no wid), + * - an execution is in progress (the live stream wins), + * - the workflow has no executions or the run left no statistics. + */ + public restoreLatestRunStatistics(): Observable<void> { + if (loadPersistedHeatmapView() === null) { + return EMPTY; + } + const wid = this.workflowActionService.getWorkflowMetadata()?.wid; + if (wid === undefined) { + return EMPTY; + } + if (!isNotInExecution(this.executeWorkflowService.getExecutionState().state)) { + return EMPTY; + } + + return this.workflowExecutionsService.retrieveWorkflowExecutions(wid).pipe( + switchMap(executions => { + const run = this.pickLatestRun(executions); + if (run === undefined) { + return EMPTY; + } + return this.workflowExecutionsService.retrieveWorkflowRuntimeStatistics(wid, run.eId, run.cuId); + }), + map(rows => { Review Comment: **Polish:** This `map` transforms nothing and exists purely for the side effect of calling `setExternalStatus`. `tap` says that at a glance, and keeps the returned value from reading as meaningful. ########## frontend/src/app/workspace/service/heatmap/heatmap-stats-restore.service.spec.ts: ########## @@ -0,0 +1,208 @@ +/** Review Comment: **Advisory:** Anchored here because it is about the description's testing section. A `TODO` comment asking for a before/after GIF still opens "How was this PR tested?", but the GIF is already there: two videos in section 1, plus the click path. It is invisible in rendered Markdown, which is how it survived, and it lands in the permanent record. ########## frontend/src/app/workspace/service/heatmap/heatmap-stats-restore.service.spec.ts: ########## @@ -0,0 +1,208 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { TestBed } from "@angular/core/testing"; +import { of, throwError } from "rxjs"; +import type { Mocked } from "vitest"; +import { HeatmapStatsRestoreService } from "./heatmap-stats-restore.service"; +import { savePersistedHeatmapView } from "./heatmap-overlay-persistence"; +import { HeatmapView } from "./heatmap-scoring"; +import { WorkflowExecutionsService } from "../../../dashboard/service/user/workflow-executions/workflow-executions.service"; +import { WorkflowActionService } from "../workflow-graph/model/workflow-action.service"; +import { WorkflowStatusService } from "../workflow-status/workflow-status.service"; +import { ExecuteWorkflowService } from "../execute-workflow/execute-workflow.service"; +import { ExecutionState, OperatorState } from "../../types/execute-workflow.interface"; +import { WorkflowExecutionsEntry } from "../../../dashboard/type/workflow-executions-entry"; +import { WorkflowRuntimeStatistics } from "../../../dashboard/type/workflow-runtime-statistics"; + +function makeExecution(overrides: Partial<WorkflowExecutionsEntry>): WorkflowExecutionsEntry { + return { + eId: 1, + vId: 1, + cuId: 7, + sId: 0, + userName: "user", + avatar: "", + name: "run", + startingTime: 1_000, + completionTime: 2_000, + status: 3, // Completed + result: "", + bookmarked: false, + logLocation: "", + ...overrides, + }; +} + +function makeStatsRow(overrides: Partial<WorkflowRuntimeStatistics>): WorkflowRuntimeStatistics { + return { + operatorId: "op1", + timestamp: 1_000, + inputTupleCount: 10, + inputTupleSize: 100, + outputTupleCount: 5, + outputTupleSize: 50, + totalDataProcessingTime: 1_000_000, + totalControlProcessingTime: 2_000, + totalIdleTime: 3_000, + numberOfWorkers: 2, + status: 3, + ...overrides, + }; +} + +describe("HeatmapStatsRestoreService", () => { + let service: HeatmapStatsRestoreService; + let executionsService: Mocked<WorkflowExecutionsService>; + let statusService: Mocked<WorkflowStatusService>; + let actionService: Mocked<WorkflowActionService>; + let executeService: Mocked<ExecuteWorkflowService>; + + beforeEach(() => { + localStorage.clear(); + // Default arrangement: overlay persisted on, a saved workflow, no live run. + savePersistedHeatmapView(HeatmapView.Runtime); + + executionsService = { + retrieveWorkflowExecutions: vi.fn(() => of([makeExecution({})])), + retrieveWorkflowRuntimeStatistics: vi.fn(() => of([makeStatsRow({})])), + } as unknown as Mocked<WorkflowExecutionsService>; + actionService = { + getWorkflowMetadata: vi.fn(() => ({ wid: 42 })), + } as unknown as Mocked<WorkflowActionService>; + statusService = { setExternalStatus: vi.fn() } as unknown as Mocked<WorkflowStatusService>; + executeService = { + getExecutionState: vi.fn(() => ({ state: ExecutionState.Uninitialized })), + } as unknown as Mocked<ExecuteWorkflowService>; + + TestBed.configureTestingModule({ + providers: [ + HeatmapStatsRestoreService, + { provide: WorkflowExecutionsService, useValue: executionsService }, + { provide: WorkflowActionService, useValue: actionService }, + { provide: WorkflowStatusService, useValue: statusService }, + { provide: ExecuteWorkflowService, useValue: executeService }, + ], + }); + service = TestBed.inject(HeatmapStatsRestoreService); + }); + + afterEach(() => localStorage.clear()); + + it("fetches the latest run and feeds the mapped statistics into WorkflowStatusService", () => { + service.restoreLatestRunStatistics().subscribe(); + + expect(executionsService.retrieveWorkflowExecutions).toHaveBeenCalledWith(42); + expect(executionsService.retrieveWorkflowRuntimeStatistics).toHaveBeenCalledWith(42, 1, 7); + expect(statusService.setExternalStatus).toHaveBeenCalledWith({ + op1: expect.objectContaining({ + operatorState: OperatorState.Completed, + aggregatedInputRowCount: 10, + aggregatedOutputRowCount: 5, + aggregatedDataProcessingTime: 1_000_000, + numWorkers: 2, + }), + }); + }); + + it("does nothing when the overlay is not persisted on", () => { + savePersistedHeatmapView(null); + + service.restoreLatestRunStatistics().subscribe(); + + expect(executionsService.retrieveWorkflowExecutions).not.toHaveBeenCalled(); + expect(statusService.setExternalStatus).not.toHaveBeenCalled(); + }); + + it("does nothing for an unsaved workflow (no wid)", () => { + actionService.getWorkflowMetadata.mockReturnValue({ wid: undefined } as never); + + service.restoreLatestRunStatistics().subscribe(); + + expect(executionsService.retrieveWorkflowExecutions).not.toHaveBeenCalled(); + expect(statusService.setExternalStatus).not.toHaveBeenCalled(); + }); + + it("skips the restore while an execution is in progress, so the live stream wins", () => { + executeService.getExecutionState.mockReturnValue({ state: ExecutionState.Running } as never); + + service.restoreLatestRunStatistics().subscribe(); + + expect(executionsService.retrieveWorkflowExecutions).not.toHaveBeenCalled(); + expect(statusService.setExternalStatus).not.toHaveBeenCalled(); + }); + + it("neither ingests nor throws when the workflow has no executions", () => { + executionsService.retrieveWorkflowExecutions.mockReturnValue(of([])); + + expect(() => service.restoreLatestRunStatistics().subscribe()).not.toThrow(); + + expect(executionsService.retrieveWorkflowRuntimeStatistics).not.toHaveBeenCalled(); + expect(statusService.setExternalStatus).not.toHaveBeenCalled(); + }); + + it("prefers the most recent completed run over newer unfinished ones", () => { + executionsService.retrieveWorkflowExecutions.mockReturnValue( + of([ + makeExecution({ eId: 3, cuId: 9, startingTime: 3_000, status: 4 }), // newest, Failed + makeExecution({ eId: 2, cuId: 8, startingTime: 2_000, status: 3 }), // newest Completed + makeExecution({ eId: 1, cuId: 7, startingTime: 1_000, status: 3 }), + ]) + ); + + service.restoreLatestRunStatistics().subscribe(); + + expect(executionsService.retrieveWorkflowRuntimeStatistics).toHaveBeenCalledWith(42, 2, 8); + }); + + it("still loads the latest run when no execution ever completed", () => { + executionsService.retrieveWorkflowExecutions.mockReturnValue( + of([ + makeExecution({ eId: 5, cuId: 11, startingTime: 5_000, status: 1 }), // Running + makeExecution({ eId: 4, cuId: 10, startingTime: 4_000, status: 4 }), // Failed + ]) + ); + + service.restoreLatestRunStatistics().subscribe(); + + expect(executionsService.retrieveWorkflowRuntimeStatistics).toHaveBeenCalledWith(42, 5, 11); + }); + + it("swallows an HTTP error from the executions fetch", () => { Review Comment: **Advisory:** Both specs assert only `not.toThrow()` and `setExternalStatus` not called. Both hold with `catchError` deleted: the error short-circuits the `map` either way, and RxJS 7 reports an unhandled error asynchronously rather than rethrowing from `subscribe()` (7.8.2, `frontend/package.json:69`). Subscribing with an explicit `{ error, complete }` pair would pin the behaviour you mean. I could not run the suite to settle this, so please treat it as a question. ########## frontend/src/app/workspace/service/heatmap/heatmap-stats-restore.service.ts: ########## @@ -0,0 +1,105 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { Injectable } from "@angular/core"; +import { EMPTY, Observable, of } from "rxjs"; +import { catchError, map, switchMap } from "rxjs/operators"; +import { WorkflowExecutionsService } from "../../../dashboard/service/user/workflow-executions/workflow-executions.service"; +import { WorkflowExecutionsEntry } from "../../../dashboard/type/workflow-executions-entry"; +import { WorkflowActionService } from "../workflow-graph/model/workflow-action.service"; +import { WorkflowStatusService } from "../workflow-status/workflow-status.service"; +import { ExecuteWorkflowService } from "../execute-workflow/execute-workflow.service"; +import { isNotInExecution } from "../../types/execute-workflow.interface"; +import { loadPersistedHeatmapView } from "./heatmap-overlay-persistence"; +import { toOperatorRuntimeStatusMap } from "./runtime-statistics-mapper"; + +/** Persisted execution status code for a completed run (EXECUTION_STATUS_CODE). */ +const EXECUTION_COMPLETED_CODE = 3; + +/** + * Restores the last execution's per-operator statistics after a page refresh, + * so the performance heat-map can render a finished run without re-executing. + * + * All gating lives here rather than in the caller: the persisted-overlay check + * is a synchronous localStorage read, so a user who never enabled the overlay + * incurs no fetch at all. + */ +@Injectable({ + providedIn: "root", +}) +export class HeatmapStatsRestoreService { + constructor( + private workflowExecutionsService: WorkflowExecutionsService, + private workflowActionService: WorkflowActionService, + private workflowStatusService: WorkflowStatusService, + private executeWorkflowService: ExecuteWorkflowService + ) {} + + /** + * Fetches the latest run's statistics and feeds them into + * WorkflowStatusService. Cold: nothing happens until subscribed. Skips Review Comment: **Polish:** "Cold: nothing happens until subscribed" is not quite true of the gates listed below it. All three run eagerly at `:64`, `:67` and `:71`; only the fetches defer. Worth fixing separately from the ordering question, since the prose and the race are independent. ########## frontend/src/app/workspace/service/heatmap/heatmap-stats-restore.service.ts: ########## @@ -0,0 +1,105 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { Injectable } from "@angular/core"; +import { EMPTY, Observable, of } from "rxjs"; +import { catchError, map, switchMap } from "rxjs/operators"; +import { WorkflowExecutionsService } from "../../../dashboard/service/user/workflow-executions/workflow-executions.service"; +import { WorkflowExecutionsEntry } from "../../../dashboard/type/workflow-executions-entry"; +import { WorkflowActionService } from "../workflow-graph/model/workflow-action.service"; +import { WorkflowStatusService } from "../workflow-status/workflow-status.service"; +import { ExecuteWorkflowService } from "../execute-workflow/execute-workflow.service"; +import { isNotInExecution } from "../../types/execute-workflow.interface"; +import { loadPersistedHeatmapView } from "./heatmap-overlay-persistence"; +import { toOperatorRuntimeStatusMap } from "./runtime-statistics-mapper"; + +/** Persisted execution status code for a completed run (EXECUTION_STATUS_CODE). */ +const EXECUTION_COMPLETED_CODE = 3; + +/** + * Restores the last execution's per-operator statistics after a page refresh, + * so the performance heat-map can render a finished run without re-executing. + * + * All gating lives here rather than in the caller: the persisted-overlay check + * is a synchronous localStorage read, so a user who never enabled the overlay + * incurs no fetch at all. + */ +@Injectable({ + providedIn: "root", +}) +export class HeatmapStatsRestoreService { + constructor( + private workflowExecutionsService: WorkflowExecutionsService, + private workflowActionService: WorkflowActionService, + private workflowStatusService: WorkflowStatusService, + private executeWorkflowService: ExecuteWorkflowService + ) {} + + /** + * Fetches the latest run's statistics and feeds them into + * WorkflowStatusService. Cold: nothing happens until subscribed. Skips + * silently (including on HTTP errors — restoring is best-effort) when: + * - the overlay is not persisted on, + * - the workflow has never been saved (no wid), + * - an execution is in progress (the live stream wins), + * - the workflow has no executions or the run left no statistics. + */ + public restoreLatestRunStatistics(): Observable<void> { + if (loadPersistedHeatmapView() === null) { + return EMPTY; + } + const wid = this.workflowActionService.getWorkflowMetadata()?.wid; + if (wid === undefined) { + return EMPTY; + } + if (!isNotInExecution(this.executeWorkflowService.getExecutionState().state)) { + return EMPTY; + } + + return this.workflowExecutionsService.retrieveWorkflowExecutions(wid).pipe( + switchMap(executions => { + const run = this.pickLatestRun(executions); + if (run === undefined) { + return EMPTY; + } + return this.workflowExecutionsService.retrieveWorkflowRuntimeStatistics(wid, run.eId, run.cuId); + }), + map(rows => { + const runtimeStatus = toOperatorRuntimeStatusMap(rows); + if (Object.keys(runtimeStatus).length > 0) { + this.workflowStatusService.setExternalStatus(runtimeStatus); + } + }), + catchError(() => EMPTY) Review Comment: **Advisory:** Best-effort is right for the HTTP legs. The placement is the issue: sitting after the `map`, this also swallows a mapper or ingestion throw, and nothing is logged, so a failed restore looks like a correctly skipped one. `computing-unit-selection.component.ts:328-333` has a real `error:` handler with a fallback. Moving `catchError` above the `map` keeps the HTTP legs best-effort. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
