Yicong-Huang commented on code in PR #5834: URL: https://github.com/apache/texera/pull/5834#discussion_r3478227197
########## frontend/src/app/workspace/service/workflow-status/performance-metrics.spec.ts: ########## @@ -0,0 +1,116 @@ +/** + * 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 { OperatorPerformanceMetrics, toPerformanceMetrics } from "./performance-metrics"; +import { OperatorState, OperatorStatistics } from "../../types/execute-workflow.interface"; + +/** + * A complete statistics object, mirroring what the backend sends once every + * field is typed. All five timing/size fields are present. + */ +const fullStats: OperatorStatistics = { + operatorState: OperatorState.Running, + aggregatedInputRowCount: 1_000_000, + aggregatedInputSize: 84_000_000, + inputPortMetrics: { "0": 1_000_000 }, + aggregatedOutputRowCount: 12_000, + aggregatedOutputSize: 1_010_000, + outputPortMetrics: { "0": 12_000 }, + numWorkers: 4, + aggregatedDataProcessingTime: 8_500_000_000, + aggregatedControlProcessingTime: 120_000_000, + aggregatedIdleTime: 300_000_000, +}; + +/** + * The partial shape produced by WorkflowStatusService.resetStatus(): only the + * required fields, none of the five optional timing/size fields, no numWorkers. + * The mapper must survive this without emitting NaN/undefined. + */ +const partialStats: OperatorStatistics = { + operatorState: OperatorState.Uninitialized, + aggregatedInputRowCount: 0, + inputPortMetrics: {}, + aggregatedOutputRowCount: 0, + outputPortMetrics: {}, +}; + +describe("toPerformanceMetrics", () => { + it("maps every field from a full statistics object", () => { + const m: OperatorPerformanceMetrics = toPerformanceMetrics("Filter-operator-a1b2", fullStats); + expect(m).toEqual({ + operatorId: "Filter-operator-a1b2", + dataProcessingTimeNs: 8_500_000_000, + controlProcessingTimeNs: 120_000_000, + idleTimeNs: 300_000_000, + inputRows: 1_000_000, + outputRows: 12_000, + inputSize: 84_000_000, + outputSize: 1_010_000, + numWorkers: 4, + }); + }); + + it("keeps data and control processing time as separate fields", () => { + const m = toPerformanceMetrics("op", fullStats); + expect(m.dataProcessingTimeNs).toBe(fullStats.aggregatedDataProcessingTime); + expect(m.controlProcessingTimeNs).toBe(fullStats.aggregatedControlProcessingTime); + }); + + it("defaults every optional/missing field to 0 (no NaN, no undefined)", () => { + const m = toPerformanceMetrics("op", partialStats); + expect(m).toEqual({ + operatorId: "op", + dataProcessingTimeNs: 0, + controlProcessingTimeNs: 0, + idleTimeNs: 0, + inputRows: 0, + outputRows: 0, + inputSize: 0, + outputSize: 0, + numWorkers: 0, Review Comment: numWorkers can never be 0, it is weird to default to 0. ########## frontend/src/app/workspace/service/workflow-status/performance-metrics.ts: ########## @@ -0,0 +1,69 @@ +/** + * 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 { OperatorStatistics } from "../../types/execute-workflow.interface"; + +/** + * Derived per-operator performance metrics. + * + * This is the ground-truth model captured by {@link WorkflowStatusService}. It is + * a flat, defensively-defaulted projection of the raw {@link OperatorStatistics} + * the backend streams over the websocket — every field is a finite, non-negative + * number, so downstream consumers never have to re-validate. + */ +export interface OperatorPerformanceMetrics + extends Readonly<{ + operatorId: string; Review Comment: maybe this should be a key: value pairs? operator should be the key? ########## frontend/src/app/workspace/service/workflow-status/workflow-status.service.ts: ########## @@ -49,6 +60,35 @@ export class WorkflowStatusService { return this.currentStatus; } + /** Stream of derived per-operator performance metrics, keyed by operator id. */ + public getPerformanceMetricsStream(): Observable<Record<string, OperatorPerformanceMetrics>> { + return this.performanceMetricsSubject.asObservable(); + } + + /** Synchronous snapshot of the latest derived per-operator performance metrics. */ + public getCurrentPerformanceMetrics(): Record<string, OperatorPerformanceMetrics> { + return this.performanceMetricsSubject.getValue(); + } + + /** + * Feed externally-sourced statistics (e.g. restored historical runtime stats) + * through the same pipeline as live websocket updates, so derived metrics and + * all subscribers update identically. + */ + public setExternalStatus(status: Record<string, OperatorStatistics>): void { + this.statusSubject.next(status); + } Review Comment: I don't understand what is external status? how are they generated and how are they used? ########## frontend/src/app/workspace/service/workflow-status/performance-metrics.ts: ########## @@ -0,0 +1,69 @@ +/** + * 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 { OperatorStatistics } from "../../types/execute-workflow.interface"; + +/** + * Derived per-operator performance metrics. + * + * This is the ground-truth model captured by {@link WorkflowStatusService}. It is + * a flat, defensively-defaulted projection of the raw {@link OperatorStatistics} + * the backend streams over the websocket — every field is a finite, non-negative + * number, so downstream consumers never have to re-validate. + */ +export interface OperatorPerformanceMetrics + extends Readonly<{ + operatorId: string; + dataProcessingTimeNs: number; + controlProcessingTimeNs: number; + idleTimeNs: number; + inputRows: number; + outputRows: number; + inputSize: number; + outputSize: number; + numWorkers: number; + }> {} + +/** + * Coerce an untrusted numeric field (it arrives over the websocket) into a + * finite, non-negative number. Anything missing, non-numeric, NaN, infinite, or + * negative collapses to 0 so no NaN/Infinity can leak into downstream consumers. + */ +function toFiniteNonNegative(value: number | undefined): number { + return typeof value === "number" && Number.isFinite(value) && value > 0 ? value : 0; +} + +/** + * Project a single raw {@link OperatorStatistics} into the flat performance model, + * defaulting every optional/missing field to 0. Data and control processing time + * are kept as separate fields so consumers can choose how to combine them. + */ +export function toPerformanceMetrics(operatorId: string, stats: OperatorStatistics): OperatorPerformanceMetrics { Review Comment: rename to `extractPerformanceMetrics`? ########## frontend/src/app/workspace/service/workflow-status/performance-metrics.ts: ########## @@ -0,0 +1,69 @@ +/** + * 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 { OperatorStatistics } from "../../types/execute-workflow.interface"; + +/** + * Derived per-operator performance metrics. + * + * This is the ground-truth model captured by {@link WorkflowStatusService}. It is + * a flat, defensively-defaulted projection of the raw {@link OperatorStatistics} + * the backend streams over the websocket — every field is a finite, non-negative + * number, so downstream consumers never have to re-validate. + */ +export interface OperatorPerformanceMetrics + extends Readonly<{ + operatorId: string; + dataProcessingTimeNs: number; + controlProcessingTimeNs: number; + idleTimeNs: number; + inputRows: number; + outputRows: number; + inputSize: number; + outputSize: number; + numWorkers: number; + }> {} + +/** + * Coerce an untrusted numeric field (it arrives over the websocket) into a + * finite, non-negative number. Anything missing, non-numeric, NaN, infinite, or + * negative collapses to 0 so no NaN/Infinity can leak into downstream consumers. + */ Review Comment: it is good to be defensive, but are they any cases where our backend is sending malformated data? -- 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]
