From d35d3a24d071cac05c79be91ac86653aa8497722 Mon Sep 17 00:00:00 2001 From: kary zheng Date: Sun, 4 Oct 2026 00:35:18 -0700 Subject: [PATCH 1/8] fix(frontend): dock the result panel at its default height The default height went from 300 to 500 while the docked offset stayed at -300px, so a panel that had never been moved, and every press of the dock button, left 200px of it below the window. The offset now comes from the same constant. Co-Authored-By: Claude Opus 5.5 --- .../component/result-panel/result-panel.component.html | 1 + .../component/result-panel/result-panel.component.scss | 1 - .../component/result-panel/result-panel.component.spec.ts | 7 +++++++ .../component/result-panel/result-panel.component.ts | 3 +++ 4 files changed, 11 insertions(+), 1 deletion(-) diff --git a/frontend/src/app/workspace/component/result-panel/result-panel.component.html b/frontend/src/app/workspace/component/result-panel/result-panel.component.html index d2c6a535493..5d5d7009c6e 100644 --- a/frontend/src/app/workspace/component/result-panel/result-panel.component.html +++ b/frontend/src/app/workspace/component/result-panel/result-panel.component.html @@ -55,6 +55,7 @@ [nzMinHeight]="250" [nzMaxWidth]="window.innerWidth" [nzMaxHeight]="window.innerHeight" + [style.top.px]="-defaultHeight" [style.width.px]="width" [style.height.px]="height" (nzResize)="onResize($event)" diff --git a/frontend/src/app/workspace/component/result-panel/result-panel.component.scss b/frontend/src/app/workspace/component/result-panel/result-panel.component.scss index 229857b4a44..17f54bd89b0 100644 --- a/frontend/src/app/workspace/component/result-panel/result-panel.component.scss +++ b/frontend/src/app/workspace/component/result-panel/result-panel.component.scss @@ -31,7 +31,6 @@ #result-container { position: absolute; - top: -300px; left: 0; background: white; border-radius: 5px; diff --git a/frontend/src/app/workspace/component/result-panel/result-panel.component.spec.ts b/frontend/src/app/workspace/component/result-panel/result-panel.component.spec.ts index abe5fe1eb2a..7608f7a2ba1 100644 --- a/frontend/src/app/workspace/component/result-panel/result-panel.component.spec.ts +++ b/frontend/src/app/workspace/component/result-panel/result-panel.component.spec.ts @@ -692,6 +692,13 @@ describe("ResultPanelComponent", () => { el.triggerEventHandler(event, payload); }; + // The return position is computed against the default height. Docked a different distance + // above the workspace's edge, a panel opened at that height runs off the bottom of the window. + it("docks the panel one default height above the workspace's bottom edge", () => { + const container: HTMLElement = fixture.debugElement.query(By.css("#result-container")).nativeElement; + expect(container.style.top).toBe(`-${DEFAULT_HEIGHT}px`); + }); + it("renders the docked panel chrome and wires the close / reset actions", () => { component.width = DEFAULT_WIDTH; fixture.detectChanges(); diff --git a/frontend/src/app/workspace/component/result-panel/result-panel.component.ts b/frontend/src/app/workspace/component/result-panel/result-panel.component.ts index 88c29eed760..38e9555c1f4 100644 --- a/frontend/src/app/workspace/component/result-panel/result-panel.component.ts +++ b/frontend/src/app/workspace/component/result-panel/result-panel.component.ts @@ -97,6 +97,9 @@ export class ResultPanelComponent implements OnInit, OnDestroy { componentOutlets!: ElementRef; frameComponentConfigs: Map; componentInputs: {} }> = new Map(); protected readonly window = window; + // The docked panel sits this far above the workspace's bottom edge. The return position is + // computed against the default height, so the two have to be the same number. + protected readonly defaultHeight = DEFAULT_HEIGHT; id = -1; width = DEFAULT_WIDTH; height = DEFAULT_HEIGHT; From 88467568316f226e12746d812a4601c0cecd3fac Mon Sep 17 00:00:00 2001 From: kary zheng Date: Tue, 20 Oct 2026 16:03:10 -0700 Subject: [PATCH 2/8] feat: read the result of any output port of an operator Every external output port of an operator that keeps its result is stored, but results were only ever read from the first port, so a Split's second output or an If's True branch could not be shown. A pagination request can now name the output port to read, with the first port as the default, and the reply carries the port it read and that port's row count. The live result updates keep following the first port, so a client paging another port takes the count from the reply. The result store's count per operator now comes from the first port too; it was built from every stored port, so a terminal Split's second output could replace its first output's count. On the frontend, the result service gives each extra port its own pagination service, which names its port in its requests, keeps no page cache and takes its row count from each reply, and every port's service takes its schema only from replies to its own requests. The result table and its row detail can show any port. Co-Authored-By: Claude Opus 5.5 --- .../event/PaginatedResultEvent.scala | 21 +++- .../request/ResultPaginationRequest.scala | 4 +- .../web/service/ExecutionResultService.scala | 14 ++- .../event/PaginatedResultEventSpec.scala | 24 ++++- .../event/TexeraWebSocketEventSpec.scala | 4 +- .../WorkflowWebsocketResourceSpec.scala | 4 +- .../service/ExecutionResultServiceSpec.scala | 71 ++++++++++-- .../result-panel-modal.component.spec.ts | 14 ++- .../result-panel-modal.component.ts | 12 ++- .../result-table-frame.component.spec.ts | 38 ++++++- .../result-table-frame.component.ts | 29 +++-- .../workflow-result.service.spec.ts | 101 +++++++++++++++++- .../workflow-result.service.ts | 46 +++++++- .../types/workflow-websocket.interface.ts | 5 + 14 files changed, 349 insertions(+), 38 deletions(-) diff --git a/amber/src/main/scala/org/apache/texera/web/model/websocket/event/PaginatedResultEvent.scala b/amber/src/main/scala/org/apache/texera/web/model/websocket/event/PaginatedResultEvent.scala index 081d072180b..9f7983241e3 100644 --- a/amber/src/main/scala/org/apache/texera/web/model/websocket/event/PaginatedResultEvent.scala +++ b/amber/src/main/scala/org/apache/texera/web/model/websocket/event/PaginatedResultEvent.scala @@ -27,16 +27,31 @@ object PaginatedResultEvent { def apply( req: ResultPaginationRequest, table: List[ObjectNode], - schema: List[Attribute] + schema: List[Attribute], + totalTuples: Int = 0 ): PaginatedResultEvent = { - PaginatedResultEvent(req.requestID, req.operatorID, req.pageIndex, table, schema) + PaginatedResultEvent( + req.requestID, + req.operatorID, + req.pageIndex, + table, + schema, + req.portIndex, + totalTuples + ) } } +/** + * One page of an output port's result. `totalTuples` counts the whole port, so a port that the + * running execution does not report on, any but an operator's first, can still be paged through. + */ case class PaginatedResultEvent( requestID: String, operatorID: String, pageIndex: Int, table: List[ObjectNode], - schema: List[Attribute] + schema: List[Attribute], + portIndex: Int, + totalTuples: Int ) extends TexeraWebSocketEvent diff --git a/amber/src/main/scala/org/apache/texera/web/model/websocket/request/ResultPaginationRequest.scala b/amber/src/main/scala/org/apache/texera/web/model/websocket/request/ResultPaginationRequest.scala index 4a3e0a58a3e..2696d90bcb3 100644 --- a/amber/src/main/scala/org/apache/texera/web/model/websocket/request/ResultPaginationRequest.scala +++ b/amber/src/main/scala/org/apache/texera/web/model/websocket/request/ResultPaginationRequest.scala @@ -26,5 +26,7 @@ case class ResultPaginationRequest( pageSize: Int, columnOffset: Int = 0, columnLimit: Int = Int.MaxValue, - columnSearch: Option[String] = None + columnSearch: Option[String] = None, + // which of the operator's output ports to read; an operator with one output has only port 0 + portIndex: Int = 0 ) extends TexeraWebSocketRequest diff --git a/amber/src/main/scala/org/apache/texera/web/service/ExecutionResultService.scala b/amber/src/main/scala/org/apache/texera/web/service/ExecutionResultService.scala index 2fa63f6b863..50d5c03de6b 100644 --- a/amber/src/main/scala/org/apache/texera/web/service/ExecutionResultService.scala +++ b/amber/src/main/scala/org/apache/texera/web/service/ExecutionResultService.scala @@ -478,7 +478,7 @@ class ExecutionResultService( val storageUriOption = WorkflowExecutionsResource.getResultUriByLogicalPortId( latestExecutionId, OperatorIdentity(request.operatorID), - PortIdentity() + PortIdentity(request.portIndex) ) // Refuse to read a per-user-warehouse result while the feature is off (#6930). @@ -511,7 +511,12 @@ class ExecutionResultService( val attributes = paginationIterable.headOption .map(_.getSchema.getAttributes) .getOrElse(List.empty) - PaginatedResultEvent.apply(request, mappedResults, attributes) + PaginatedResultEvent.apply( + request, + mappedResults, + attributes, + virtualDocument.getCount.toInt + ) case None => // Handle the case when storageUri is empty @@ -524,6 +529,11 @@ class ExecutionResultService( val newInfo: Map[OperatorIdentity, OperatorResultMetadata] = { WorkflowExecutionsResource .getResultUrisByExecutionId(executionId) + // One entry per operator, so it follows the first output port, as the result updates do. + // Keyed by operator, a second port's count would otherwise replace the first's. + .filter(uri => + VFSURIFactory.decodeURI(uri).globalPortId.exists(_.portId == PortIdentity()) + ) .map(uri => { val count = DocumentFactory.openDocument(uri)._1.getCount.toInt diff --git a/amber/src/test/scala/org/apache/texera/web/model/websocket/event/PaginatedResultEventSpec.scala b/amber/src/test/scala/org/apache/texera/web/model/websocket/event/PaginatedResultEventSpec.scala index aa48cc81c8c..d07c881fa74 100644 --- a/amber/src/test/scala/org/apache/texera/web/model/websocket/event/PaginatedResultEventSpec.scala +++ b/amber/src/test/scala/org/apache/texera/web/model/websocket/event/PaginatedResultEventSpec.scala @@ -60,7 +60,8 @@ class PaginatedResultEventSpec extends AnyFlatSpec with Matchers { pageSize = 25, columnOffset = 4, columnLimit = 6, - columnSearch = Some("city") + columnSearch = Some("city"), + portIndex = 1 ) private val table = List( @@ -80,6 +81,12 @@ class PaginatedResultEventSpec extends AnyFlatSpec with Matchers { event.pageIndex shouldBe 3 } + it should "echo the port the request read, and the port's row count" in { + val event = PaginatedResultEvent(request, table, schema, totalTuples = 40) + event.portIndex shouldBe 1 + event.totalTuples shouldBe 40 + } + it should "pass the table and schema through untouched" in { val event = PaginatedResultEvent(request, table, schema) event.table should contain theSameElementsInOrderAs table @@ -88,10 +95,21 @@ class PaginatedResultEventSpec extends AnyFlatSpec with Matchers { "the projected event" should "serialize to exactly the fields the Angular client declares" in { val json = objectMapper.readTree( - objectMapper.writeValueAsString(PaginatedResultEvent(request, table, schema)) + objectMapper.writeValueAsString(PaginatedResultEvent(request, table, schema, 40)) ) json.fieldNames().asScala.toSet shouldBe - Set("type", "requestID", "operatorID", "pageIndex", "table", "schema") + Set( + "type", + "requestID", + "operatorID", + "pageIndex", + "table", + "schema", + "portIndex", + "totalTuples" + ) + json.get("portIndex").asInt() shouldBe 1 + json.get("totalTuples").asInt() shouldBe 40 json.get("type").asText() shouldBe "PaginatedResultEvent" json.get("requestID").asText() shouldBe "req-alpha" json.get("operatorID").asText() shouldBe "op-beta" diff --git a/amber/src/test/scala/org/apache/texera/web/model/websocket/event/TexeraWebSocketEventSpec.scala b/amber/src/test/scala/org/apache/texera/web/model/websocket/event/TexeraWebSocketEventSpec.scala index 19dd3af6b7c..2aa250b3a7c 100644 --- a/amber/src/test/scala/org/apache/texera/web/model/websocket/event/TexeraWebSocketEventSpec.scala +++ b/amber/src/test/scala/org/apache/texera/web/model/websocket/event/TexeraWebSocketEventSpec.scala @@ -147,7 +147,9 @@ class TexeraWebSocketEventSpec extends AnyFlatSpec with Matchers { "op-2", 3, List(resultRow), - List(new Attribute("city", AttributeType.STRING)) + List(new Attribute("city", AttributeType.STRING)), + 1, + 40 ), "PythonExpressionEvaluateResponse" -> PythonExpressionEvaluateResponse( "len(tuple_)", diff --git a/amber/src/test/scala/org/apache/texera/web/resource/WorkflowWebsocketResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/WorkflowWebsocketResourceSpec.scala index 4761c5b3022..81057d74efa 100644 --- a/amber/src/test/scala/org/apache/texera/web/resource/WorkflowWebsocketResourceSpec.scala +++ b/amber/src/test/scala/org/apache/texera/web/resource/WorkflowWebsocketResourceSpec.scala @@ -284,7 +284,9 @@ class WorkflowWebsocketResourceSpec request.operatorID, request.pageIndex, List(objectMapper.createObjectNode().put("c", 5)), - List(new Attribute("c", AttributeType.INTEGER)) + List(new Attribute("c", AttributeType.INTEGER)), + request.portIndex, + 1 ) } } diff --git a/amber/src/test/scala/org/apache/texera/web/service/ExecutionResultServiceSpec.scala b/amber/src/test/scala/org/apache/texera/web/service/ExecutionResultServiceSpec.scala index 3d9ed9fcf8a..0fec412ec5a 100644 --- a/amber/src/test/scala/org/apache/texera/web/service/ExecutionResultServiceSpec.scala +++ b/amber/src/test/scala/org/apache/texera/web/service/ExecutionResultServiceSpec.scala @@ -925,6 +925,32 @@ class ExecutionResultServiceSpec event.schema.map(_.getName) shouldBe List("c1", "c2") } + it should "read the output port the request names, and count that port's rows" in { + val executionId = newExecution() + val schema = new Schema(List(new Attribute("id", AttributeType.INTEGER))) + val rows = (n: Int, offset: Int) => + (0 until n).map(i => + Tuple.builder(schema).add("id", AttributeType.INTEGER, offset + i).build() + ) + // Two output ports of one operator, as a Split keeps them, with different row counts. + storeResult(executionId, "erss-two-ports", schema, rows(2, 0)) + storeResult(executionId, "erss-two-ports", schema, rows(5, 100), port = PortIdentity(1)) + + val first = paginate(paginationRequest.copy(operatorID = "erss-two-ports", pageSize = 10)) + val second = paginate( + paginationRequest.copy(operatorID = "erss-two-ports", pageSize = 3, portIndex = 1) + ) + + // Without a port, the first one is read, as before. + first.table.map(_.get("id").asInt()) shouldBe List(0, 1) + first.portIndex shouldBe 0 + first.totalTuples shouldBe 2 + // The second port's rows, and its own count, which pages the port past this first page. + second.table.map(_.get("id").asInt()) shouldBe List(100, 101, 102) + second.portIndex shouldBe 1 + second.totalTuples shouldBe 5 + } + it should "return an empty page when the operator has no stored result" in { newExecution() @@ -1237,6 +1263,34 @@ class ExecutionResultServiceSpec } } + it should "report an operator with two output ports by its first port's count" in { + val executionId = newExecution() + val schema = new Schema(List(new Attribute("id", AttributeType.INTEGER))) + val rows = (n: Int) => + (0 until n).map(i => Tuple.builder(schema).add("id", AttributeType.INTEGER, i).build()) + storeResult(executionId, "erss-final-two-ports", schema, rows(3)) + storeResult(executionId, "erss-final-two-ports", schema, rows(8), port = PortIdentity(1)) + val plan = planOf( + physicalOp( + executionId, + "erss-final-two-ports", + PortIdentity() -> OutputMode.SET_SNAPSHOT, + PortIdentity(1) -> OutputMode.SET_SNAPSHOT + ) + ) + + withAttached(plan, executionId) { fixture => + setPollingCancellable(fixture.service, new RecordingCancellable) + + fixture.client.fire(ExecutionStateUpdate(WorkflowAggregatedState.COMPLETED)) + + // The update for an operator reads its first port, so its count has to come from there + // too; the second port's 8 rows must not stand in for the first port's 3. + val resultInfo = fixture.workflowStateStore.resultStore.getState.resultInfo + resultInfo(OperatorIdentity("erss-final-two-ports")).tupleCount shouldBe 3 + } + } + it should "leave polling running while the execution has not terminated" in { val executionId = newExecution() withAttached(planOf(), executionId) { fixture => @@ -1450,10 +1504,13 @@ class ExecutionResultServiceSpec ExecutionIdentity(execution.getEid.longValue()) } - private def globalPortIdOf(operatorId: String): GlobalPortIdentity = + private def globalPortIdOf( + operatorId: String, + port: PortIdentity = PortIdentity() + ): GlobalPortIdentity = GlobalPortIdentity( PhysicalOpIdentity(OperatorIdentity(operatorId), "main"), - PortIdentity(), + port, input = false ) @@ -1461,13 +1518,14 @@ class ExecutionResultServiceSpec private def resultUriOf( executionId: ExecutionIdentity, operatorId: String, - warehouse: Option[String] = None + warehouse: Option[String] = None, + port: PortIdentity = PortIdentity() ): URI = VFSURIFactory.resultURI( VFSURIFactory.createPortBaseURI( WorkflowIdentity(testWid.longValue()), executionId, - globalPortIdOf(operatorId), + globalPortIdOf(operatorId, port), warehouse ) ) @@ -1492,9 +1550,10 @@ class ExecutionResultServiceSpec executionId: ExecutionIdentity, operatorId: String, schema: Schema, - tuples: Seq[Tuple] + tuples: Seq[Tuple], + port: PortIdentity = PortIdentity() ): URI = { - val uri = resultUriOf(executionId, operatorId) + val uri = resultUriOf(executionId, operatorId, port = port) val document = DocumentFactory.createDocument(uri, schema) val writer = document.writer("erss").asInstanceOf[BufferedItemWriter[Tuple]] writer.open() diff --git a/frontend/src/app/workspace/component/result-panel/result-panel-modal.component.spec.ts b/frontend/src/app/workspace/component/result-panel/result-panel-modal.component.spec.ts index 84a1eeff8e2..1251c0cd883 100644 --- a/frontend/src/app/workspace/component/result-panel/result-panel-modal.component.spec.ts +++ b/frontend/src/app/workspace/component/result-panel/result-panel-modal.component.spec.ts @@ -35,7 +35,7 @@ describe("RowModalComponent", () => { const mockTupleResult = { tuple: { id: "123", value: "test_data" } }; const workflowResultServiceSpy = { - getPaginatedResultService: vi.fn().mockReturnValue({ + getPortPaginatedResultService: vi.fn().mockReturnValue({ selectTuple: vi.fn().mockReturnValue(of(mockTupleResult)), }), }; @@ -77,6 +77,12 @@ describe("RowModalComponent", () => { expect(component.currentDisplayRowData).toEqual(mockTupleResult.tuple); }); + it("reads the row from the operator's first output port when the data names none", () => { + workflowResultServiceSpy.getPortPaginatedResultService.mockClear(); + component.ngOnChanges(); + expect(workflowResultServiceSpy.getPortPaginatedResultService).toHaveBeenCalledWith("op-1", 0); + }); + it("should use data URL directly without fetching for base64 media", () => { const dataUrl = "data:image/png;base64,abc123"; (component as any).buildRowEntries({ img: dataUrl }); @@ -158,7 +164,7 @@ describe("RowModalComponent", () => { }); it("should not update row data when tuple is null", () => { - vi.spyOn(workflowResultServiceSpy, "getPaginatedResultService").mockReturnValueOnce({ + vi.spyOn(workflowResultServiceSpy, "getPortPaginatedResultService").mockReturnValueOnce({ selectTuple: vi.fn().mockReturnValue(of({ tuple: null })), } as any); component.currentDisplayRowData = { existing: "data" }; @@ -209,7 +215,7 @@ describe("RowModalComponent (with pre-loaded rowData)", () => { { provide: NzModalRef, useValue: { getConfig: () => ({}), close: vi.fn() } }, { provide: WorkflowResultService, - useValue: { getPaginatedResultService: vi.fn().mockReturnValue(null) }, + useValue: { getPortPaginatedResultService: vi.fn().mockReturnValue(null) }, }, { provide: PanelResizeService, useValue: { pageSize: 10 } }, ], @@ -261,7 +267,7 @@ describe("RowModalComponent (template rendering)", () => { { provide: NzModalRef, useValue: { getConfig: () => ({}), close: vi.fn() } }, { provide: WorkflowResultService, - useValue: { getPaginatedResultService: vi.fn().mockReturnValue(null) }, + useValue: { getPortPaginatedResultService: vi.fn().mockReturnValue(null) }, }, { provide: PanelResizeService, useValue: { pageSize: 10 } }, ], diff --git a/frontend/src/app/workspace/component/result-panel/result-panel-modal.component.ts b/frontend/src/app/workspace/component/result-panel/result-panel-modal.component.ts index dddb30caf2d..5105bcbbd0c 100644 --- a/frontend/src/app/workspace/component/result-panel/result-panel-modal.component.ts +++ b/frontend/src/app/workspace/component/result-panel/result-panel-modal.component.ts @@ -55,9 +55,15 @@ export class RowModalComponent implements OnChanges, OnDestroy { []; private readonly allocatedBlobUrls: string[] = []; // Index of current displayed row in currentResult - private readonly modalData: { operatorId: string; rowIndex: number; rowData?: Record } = - inject(NZ_MODAL_DATA); + private readonly modalData: { + operatorId: string; + rowIndex: number; + rowData?: Record; + portIndex?: number; + } = inject(NZ_MODAL_DATA); readonly operatorId: string = this.modalData.operatorId; + // the output port the row is from; left out, the first + readonly portIndex: number = this.modalData.portIndex ?? 0; rowIndex: number = this.modalData.rowIndex; currentDisplayRowData: Record = {}; @@ -88,7 +94,7 @@ export class RowModalComponent implements OnChanges, OnDestroy { ngOnChanges(): void { this.workflowResultService - .getPaginatedResultService(this.operatorId) + .getPortPaginatedResultService(this.operatorId, this.portIndex) ?.selectTuple(this.rowIndex, this.resizeService.pageSize) .pipe(untilDestroyed(this)) .subscribe(res => { diff --git a/frontend/src/app/workspace/component/result-panel/result-table-frame/result-table-frame.component.spec.ts b/frontend/src/app/workspace/component/result-panel/result-table-frame/result-table-frame.component.spec.ts index 9afb07d04c3..1a3f4d57c1a 100644 --- a/frontend/src/app/workspace/component/result-panel/result-table-frame/result-table-frame.component.spec.ts +++ b/frontend/src/app/workspace/component/result-panel/result-table-frame/result-table-frame.component.spec.ts @@ -69,6 +69,8 @@ describe("ResultTableFrameComponent", () => { pageIndex, table, schema: [], + portIndex: 0, + totalTuples: table.length, }); // Installs a paginated-result-service double behind WorkflowResultService.getPaginatedResultService. @@ -199,6 +201,28 @@ describe("ResultTableFrameComponent", () => { expect(component.tableStats).toEqual({ name: { min: 1 } }); expect(component.prevTableStats).toBe(component.tableStats); }); + + it("loads the output port it is given, and a change of port alone reloads", () => { + const secondPort = { + getCurrentTotalNumTuples: vi.fn().mockReturnValue(5), + getCurrentPageIndex: vi.fn().mockReturnValue(1), + getStats: vi.fn().mockReturnValue({}), + selectPage: vi.fn().mockReturnValue(of(makePageEvent(1))), + }; + const getPortService = vi + .spyOn(workflowResultService, "getPortPaginatedResultService") + .mockReturnValue(secondPort as unknown as OperatorPaginationResultService); + component.operatorId = "op1"; + component.portIndex = 1; + + component.ngOnChanges({ portIndex: new SimpleChange(0, 1, false) }); + + // the operator stays when only the port changes + expect(component.operatorId).toBe("op1"); + expect(getPortService).toHaveBeenCalledWith("op1", 1); + expect(secondPort.selectPage).toHaveBeenCalledTimes(1); + expect(component.currentResult).toEqual([SAMPLE_ROW]); + }); }); describe("changePaginatedResultData", () => { @@ -344,6 +368,18 @@ describe("ResultTableFrameComponent", () => { expect(component.totalNumTuples).toBe(0); }); + it("leaves a port past the first alone, as the updates are about the first port", () => { + const paginated = makePaginatedResultService(); + recreateComponent("op1"); + component.portIndex = 1; + component.totalNumTuples = 5; + + resultUpdates.next({ op1: paginationUpdate(63, [1]) }); + + expect(component.totalNumTuples).toBe(5); + expect(paginated.selectPage).not.toHaveBeenCalled(); + }); + it("ignores updates when the frame has no operator", () => { recreateComponent(undefined); @@ -579,7 +615,7 @@ describe("ResultTableFrameComponent", () => { const config: any = createSpy.mock.calls[0][0]; expect(config.nzTitle).toBe("Row Details"); expect(config.nzContent).toBe(RowModalComponent); - expect(config.nzData).toEqual({ operatorId: "op1", rowIndex: 7 }); + expect(config.nzData).toEqual({ operatorId: "op1", rowIndex: 7, portIndex: 0 }); expect(config.nzAutofocus).toBeNull(); const [prev, next, ok] = config.nzFooter; expect([prev.label, next.label, ok.label]).toEqual(["<", ">", "OK"]); diff --git a/frontend/src/app/workspace/component/result-panel/result-table-frame/result-table-frame.component.ts b/frontend/src/app/workspace/component/result-panel/result-table-frame/result-table-frame.component.ts index a329161369e..b3535fabe81 100644 --- a/frontend/src/app/workspace/component/result-panel/result-table-frame/result-table-frame.component.ts +++ b/frontend/src/app/workspace/component/result-panel/result-table-frame/result-table-frame.component.ts @@ -30,7 +30,10 @@ import { NzCellEllipsisDirective, } from "ng-zorro-antd/table"; import { WorkflowActionService } from "../../../service/workflow-graph/model/workflow-action.service"; -import { WorkflowResultService } from "../../../service/workflow-result/workflow-result.service"; +import { + OperatorPaginationResultService, + WorkflowResultService, +} from "../../../service/workflow-result/workflow-result.service"; import { PanelResizeService } from "../../../service/workflow-result/panel-resize/panel-resize.service"; import { isWebPaginationUpdate, OperatorState } from "../../../types/execute-workflow.interface"; import { IndexableObject, TableColumn } from "../../../types/result-table.interface"; @@ -88,6 +91,9 @@ export type MediaCellType = "video" | "audio" | "image" | "text"; }) export class ResultTableFrameComponent implements OnInit, OnChanges { @Input() operatorId?: string; + // which of the operator's output ports to show. The running execution reports only on the first, + // so another port is read when shown, and its counts and statistics are not followed live. + @Input() portIndex = 0; // display result table currentColumns?: TableColumn[]; @@ -135,9 +141,9 @@ export class ResultTableFrameComponent implements OnInit, OnChanges { ) {} ngOnChanges(changes: SimpleChanges): void { - this.operatorId = changes.operatorId?.currentValue; + if (changes.operatorId) this.operatorId = changes.operatorId.currentValue; if (this.operatorId) { - const paginatedResultService = this.workflowResultService.getPaginatedResultService(this.operatorId); + const paginatedResultService = this.paginatedResultService(); if (paginatedResultService) { this.isFrontPagination = false; this.totalNumTuples = paginatedResultService.getCurrentTotalNumTuples(); @@ -150,6 +156,12 @@ export class ResultTableFrameComponent implements OnInit, OnChanges { } } + /** The service that pages the port shown. */ + private paginatedResultService(): OperatorPaginationResultService | undefined { + if (!this.operatorId) return undefined; + return this.workflowResultService.getPortPaginatedResultService(this.operatorId, this.portIndex); + } + ngOnInit(): void { this.workflowStatusService .getStateUpdateStream() @@ -169,7 +181,8 @@ export class ResultTableFrameComponent implements OnInit, OnChanges { .getResultUpdateStream() .pipe(untilDestroyed(this)) .subscribe(update => { - if (!this.operatorId) { + // an update is about the operator's first port + if (!this.operatorId || this.portIndex !== 0) { return; } const opUpdate = update[this.operatorId]; @@ -190,7 +203,8 @@ export class ResultTableFrameComponent implements OnInit, OnChanges { .getResultTableStats() .pipe(untilDestroyed(this)) .subscribe(([prevStats, currentStats]) => { - if (!this.operatorId) { + // the statistics are the operator's first port's + if (!this.operatorId || this.portIndex !== 0) { return; } @@ -347,7 +361,8 @@ export class ResultTableFrameComponent implements OnInit, OnChanges { // modal title nzTitle: "Row Details", nzContent: RowModalComponent, - nzData: { operatorId: this.operatorId, rowIndex: currentRowIndex }, // set the index value and page size to the modal for navigation + // set the index value and page size to the modal for navigation + nzData: { operatorId: this.operatorId, rowIndex: currentRowIndex, portIndex: this.portIndex }, // prevent browser focusing close button (ugly square highlight) nzAutofocus: null, // modal footer buttons @@ -395,7 +410,7 @@ export class ResultTableFrameComponent implements OnInit, OnChanges { if (!this.operatorId) { return; } - const paginatedResultService = this.workflowResultService.getPaginatedResultService(this.operatorId); + const paginatedResultService = this.paginatedResultService(); if (!paginatedResultService) { return; } diff --git a/frontend/src/app/workspace/service/workflow-result/workflow-result.service.spec.ts b/frontend/src/app/workspace/service/workflow-result/workflow-result.service.spec.ts index e53de2c11fc..6bf168ed62f 100644 --- a/frontend/src/app/workspace/service/workflow-result/workflow-result.service.spec.ts +++ b/frontend/src/app/workspace/service/workflow-result/workflow-result.service.spec.ts @@ -115,6 +115,33 @@ describe("WorkflowResultService", () => { expect(updateEvents).toEqual([updates]); }); + describe("getPortPaginatedResultService", () => { + it("gives the first port's service for port 0, and nothing for an operator without a table", () => { + const ws = TestBed.inject(WorkflowWebsocketService); + pushWsEvent(ws, { type: "WebResultUpdateEvent", updates: { pagOp: paginationUpdate(3) }, tableStats: {} }); + + expect(service.getPortPaginatedResultService("pagOp", 0)).toBe(service.getPaginatedResultService("pagOp")); + expect(service.getPortPaginatedResultService("noResult", 0)).toBeUndefined(); + expect(service.getPortPaginatedResultService("noResult", 1)).toBeUndefined(); + }); + + it("makes one service for another port, kept until the results are cleared", () => { + const ws = TestBed.inject(WorkflowWebsocketService); + pushWsEvent(ws, { type: "WebResultUpdateEvent", updates: { pagOp: paginationUpdate(3) }, tableStats: {} }); + + const second = service.getPortPaginatedResultService("pagOp", 1)!; + + expect(second.portIndex).toBe(1); + expect(second.operatorID).toBe("pagOp"); + expect(second).not.toBe(service.getPaginatedResultService("pagOp")); + expect(service.getPortPaginatedResultService("pagOp", 1)).toBe(second); + + service.clearResults(); + pushWsEvent(ws, { type: "WebResultUpdateEvent", updates: { pagOp: paginationUpdate(3) }, tableStats: {} }); + expect(service.getPortPaginatedResultService("pagOp", 1)).not.toBe(second); + }); + }); + it("hasNonEmptyResult tells an empty result apart from one with rows", () => { const ws = TestBed.inject(WorkflowWebsocketService); pushWsEvent(ws, { @@ -355,6 +382,8 @@ describe("OperatorPaginationResultService", () => { pageIndex: 1, table: testTable, schema: testSchema, + portIndex: 0, + totalTuples: testTable.length, }) ); @@ -382,6 +411,8 @@ describe("OperatorPaginationResultService", () => { pageIndex: 1, table: testTable, schema: testSchema, + portIndex: 0, + totalTuples: testTable.length, }) ); @@ -407,6 +438,8 @@ describe("OperatorPaginationResultService", () => { pageIndex: 2, table: cachedPage, schema: [{ attributeName: "id", attributeType: "integer" }], + portIndex: 0, + totalTuples: 0, }); expect(service.getCurrentPageIndex()).toBe(2); }); @@ -430,6 +463,8 @@ describe("OperatorPaginationResultService", () => { pageIndex: 3, table: [{ id: 7 }], schema: [{ attributeName: "id", attributeType: "integer" }], + portIndex: 0, + totalTuples: 1, }; eventSubject.next(page); @@ -494,15 +529,75 @@ describe("OperatorPaginationResultService", () => { }); }); + describe("for an output port past the first", () => { + let secondPort: OperatorPaginationResultService; + let eventSubject: Subject; + const reply = (requestID: string, totalTuples: number): PaginatedResultEvent => ({ + requestID, + operatorID: "testOperator", + pageIndex: 1, + table: [{ id: 1 }], + schema: [{ attributeName: "id", attributeType: "integer" }], + portIndex: 1, + totalTuples, + }); + + beforeEach(() => { + secondPort = new OperatorPaginationResultService("testOperator", mockWorkflowWebsocketService, 1); + eventSubject = mockWorkflowWebsocketService.subscribeToEvent.mock.results[1] + .value as unknown as Subject; + }); + + it("names the port in its requests, and the first port's service leaves it out", () => { + secondPort.selectPage(1, 10).subscribe(); + service.selectPage(1, 10).subscribe(); + + const sent = mockWorkflowWebsocketService.send.mock.calls.map(call => call[1] as any); + expect(sent[0].portIndex).toBe(1); + expect("portIndex" in sent[1]).toBe(false); + }); + + it("takes its row count from each reply, as no live update reports on it", () => { + secondPort.selectPage(1, 10).subscribe(); + const requestID = (mockWorkflowWebsocketService.send.mock.calls[0][1] as any).requestID; + + eventSubject.next(reply(requestID, 37)); + + expect(secondPort.getCurrentTotalNumTuples()).toBe(37); + expect(secondPort.getSchema()).toEqual([{ attributeName: "id", attributeType: "integer" }]); + // the first port's service did not ask, so the reply changes nothing there + expect(service.getCurrentTotalNumTuples()).toBe(0); + expect(service.getSchema()).toEqual([]); + }); + + it("asks the server again for a page it has read, so the page is never stale", () => { + (secondPort as any).resultCache.set(1, [{ id: 0 }]); + + secondPort.selectPage(1, 10).subscribe(); + + expect(mockWorkflowWebsocketService.send).toHaveBeenCalledTimes(1); + }); + }); + describe("incoming page events", () => { - it("refreshes the schema but ignores events whose request id is unknown", () => { + // Every operator's and every port's replies arrive on one stream, so a reply to another + // service's request must not hand this one its columns. + it("ignores an event whose request id is not its own, schema included", () => { const eventSubject = mockWorkflowWebsocketService.subscribeToEvent.mock.results[0] .value as unknown as Subject; const schema: SchemaAttribute[] = [{ attributeName: "name", attributeType: "string" }]; - eventSubject.next({ requestID: "no-such-request", operatorID: "testOperator", pageIndex: 1, table: [], schema }); + eventSubject.next({ + requestID: "no-such-request", + operatorID: "testOperator", + pageIndex: 1, + table: [], + schema, + portIndex: 0, + totalTuples: 0, + }); - expect(service.getSchema()).toEqual(schema); + expect(service.getSchema()).toEqual([]); }); }); }); diff --git a/frontend/src/app/workspace/service/workflow-result/workflow-result.service.ts b/frontend/src/app/workspace/service/workflow-result/workflow-result.service.ts index e7879ba15e8..2b73e418372 100644 --- a/frontend/src/app/workspace/service/workflow-result/workflow-result.service.ts +++ b/frontend/src/app/workspace/service/workflow-result/workflow-result.service.ts @@ -43,6 +43,8 @@ import { SchemaAttribute } from "../../types/workflow-compiling.interface"; }) export class WorkflowResultService { private paginatedResultServices = new Map(); + // the services for an operator's output ports past the first, keyed "operatorID:portIndex" + private otherPortResultServices = new Map(); private operatorResultServices = new Map(); // event stream of operator result update, undefined indicates the operator result is cleared @@ -112,6 +114,26 @@ export class WorkflowResultService { return this.paginatedResultServices.get(operatorID); } + /** + * The service that pages one of the operator's output ports. The first port's is the one the + * running execution keeps up to date. Any other port is read only when asked, so its service is + * made on the first ask, for an operator whose result is a table. + */ + public getPortPaginatedResultService( + operatorID: string, + portIndex: number + ): OperatorPaginationResultService | undefined { + const firstPort = this.getPaginatedResultService(operatorID); + if (portIndex === 0 || !firstPort) return firstPort; + const key = `${operatorID}:${portIndex}`; + let service = this.otherPortResultServices.get(key); + if (!service) { + service = new OperatorPaginationResultService(operatorID, this.wsService, portIndex); + this.otherPortResultServices.set(key, service); + } + return service; + } + public getResultService(operatorID: string): OperatorResultService | undefined { return this.operatorResultServices.get(operatorID); } @@ -124,6 +146,7 @@ export class WorkflowResultService { public clearResults(): void { this.operatorResultServices.clear(); this.paginatedResultServices.clear(); + this.otherPortResultServices.clear(); this.resultTableStats.next({}); this.resultClearedStream.next(); } @@ -249,10 +272,12 @@ export class OperatorPaginationResultService { constructor( public operatorID: string, - private workflowWebsocketService: WorkflowWebsocketService + private workflowWebsocketService: WorkflowWebsocketService, + // which output port it pages. The running execution reports only on the first, so any other + // port keeps no page cache and takes its row count from each reply. + public readonly portIndex: number = 0 ) { this.workflowWebsocketService.subscribeToEvent("PaginatedResultEvent").subscribe(event => { - this.schema = event.schema; this.handlePaginationResult(event); }); } @@ -298,7 +323,8 @@ export class OperatorPaginationResultService { // update currently selected page this.currentPageIndex = pageIndex; // first fetch from frontend result cache - const useCache = columnOffset === 0 && columnLimit === Number.MAX_SAFE_INTEGER && columnSearch === ""; + const useCache = + this.portIndex === 0 && columnOffset === 0 && columnLimit === Number.MAX_SAFE_INTEGER && columnSearch === ""; const pageCache = useCache ? this.resultCache.get(pageIndex) : undefined; if (pageCache) { return of({ @@ -307,6 +333,8 @@ export class OperatorPaginationResultService { pageIndex: pageIndex, table: pageCache, schema: this.schema, + portIndex: this.portIndex, + totalTuples: this.currentTotalNumTuples, }); } else { // fetch result data from server @@ -320,6 +348,7 @@ export class OperatorPaginationResultService { columnOffset, columnLimit, columnSearch, + ...this.portField(), }); const pendingRequestSubject = new Subject(); this.pendingRequests.set(requestID, pendingRequestSubject); @@ -345,11 +374,22 @@ export class OperatorPaginationResultService { this.statsCache = statsUpdate; } + // the first port is the server's default, so a request for it leaves the field out + private portField(): { portIndex?: number } { + return this.portIndex === 0 ? {} : { portIndex: this.portIndex }; + } + private handlePaginationResult(res: PaginatedResultEvent): void { const pendingRequestSubject = this.pendingRequests.get(res.requestID); if (!pendingRequestSubject) { return; } + // Every operator's and every port's replies arrive on one stream, so only a reply to this + // service's own request may set its schema. + this.schema = res.schema; + if (this.portIndex !== 0) { + this.currentTotalNumTuples = res.totalTuples; + } pendingRequestSubject.next(res); pendingRequestSubject.complete(); this.pendingRequests.delete(res.requestID); diff --git a/frontend/src/app/workspace/types/workflow-websocket.interface.ts b/frontend/src/app/workspace/types/workflow-websocket.interface.ts index fb22a3a8a42..ba8a11a5c28 100644 --- a/frontend/src/app/workspace/types/workflow-websocket.interface.ts +++ b/frontend/src/app/workspace/types/workflow-websocket.interface.ts @@ -96,6 +96,8 @@ export type PaginationRequest = Readonly<{ columnOffset?: number; columnLimit?: number; columnSearch?: string; + // which of the operator's output ports to read; left out, the first + portIndex?: number; }>; export type PaginatedResultEvent = Readonly<{ @@ -104,6 +106,9 @@ export type PaginatedResultEvent = Readonly<{ pageIndex: number; table: ReadonlyArray; schema: ReadonlyArray; + portIndex: number; + // the port's whole row count, for a port the running execution does not report on + totalTuples: number; }>; export type ResultExportRequest = Readonly<{ From 13c1294588a30fbf216ed0db7ef4545aefb1904c Mon Sep 17 00:00:00 2001 From: kary zheng Date: Tue, 20 Oct 2026 16:37:59 -0700 Subject: [PATCH 3/8] fix(frontend): keep a result table to the output port it shows Switching the table to another output port left the previous port's rows on screen when the new port was empty, with the loading state never ending, and a late reply for the previous port could replace the new port's rows. A port change now starts from an empty table, a reply for an operator or port no longer shown is dropped, and an empty page ends the loading. Co-Authored-By: Claude Opus 5.5 --- .../result-table-frame.component.spec.ts | 55 +++++++++++++++++++ .../result-table-frame.component.ts | 21 ++++++- 2 files changed, 73 insertions(+), 3 deletions(-) diff --git a/frontend/src/app/workspace/component/result-panel/result-table-frame/result-table-frame.component.spec.ts b/frontend/src/app/workspace/component/result-panel/result-table-frame/result-table-frame.component.spec.ts index 1a3f4d57c1a..9401e29fe86 100644 --- a/frontend/src/app/workspace/component/result-panel/result-table-frame/result-table-frame.component.spec.ts +++ b/frontend/src/app/workspace/component/result-panel/result-table-frame/result-table-frame.component.spec.ts @@ -223,6 +223,61 @@ describe("ResultTableFrameComponent", () => { expect(secondPort.selectPage).toHaveBeenCalledTimes(1); expect(component.currentResult).toEqual([SAMPLE_ROW]); }); + + describe("switching between output ports", () => { + const OTHER_ROW: IndexableObject = { _id: 0, name: "bob", score: 75 }; + // a port's service whose page replies the test sends when it chooses + const portService = (total: number) => { + const replies = new Subject(); + const service = { + getCurrentTotalNumTuples: vi.fn().mockReturnValue(total), + getCurrentPageIndex: vi.fn().mockReturnValue(1), + getStats: vi.fn().mockReturnValue({}), + selectPage: vi.fn().mockReturnValue(replies), + }; + return { service, replies }; + }; + const showPort = (portIndex: number) => { + component.portIndex = portIndex; + component.ngOnChanges({ portIndex: new SimpleChange(1 - portIndex, portIndex, false) }); + }; + + it("clears the previous port's rows when the port shown is empty", () => { + const first = portService(1); + const second = portService(0); + vi.spyOn(workflowResultService, "getPortPaginatedResultService").mockImplementation( + (_, port) => (port === 0 ? first.service : second.service) as unknown as OperatorPaginationResultService + ); + component.operatorId = "op1"; + showPort(0); + first.replies.next(makePageEvent(1, [SAMPLE_ROW])); + expect(component.currentResult).toEqual([SAMPLE_ROW]); + + showPort(1); + second.replies.next(makePageEvent(1, [])); + + // the empty port shows no rows, and is no longer loading + expect(component.currentResult).toEqual([]); + expect(component.isLoadingResult).toBe(false); + }); + + it("ignores a reply from a port that is no longer shown", () => { + const first = portService(1); + const second = portService(1); + vi.spyOn(workflowResultService, "getPortPaginatedResultService").mockImplementation( + (_, port) => (port === 0 ? first.service : second.service) as unknown as OperatorPaginationResultService + ); + component.operatorId = "op1"; + showPort(0); + showPort(1); + second.replies.next(makePageEvent(1, [OTHER_ROW])); + + // port 0's page 1 arrives late, after port 1's + first.replies.next(makePageEvent(1, [SAMPLE_ROW])); + + expect(component.currentResult).toEqual([OTHER_ROW]); + }); + }); }); describe("changePaginatedResultData", () => { diff --git a/frontend/src/app/workspace/component/result-panel/result-table-frame/result-table-frame.component.ts b/frontend/src/app/workspace/component/result-panel/result-table-frame/result-table-frame.component.ts index b3535fabe81..cca6c04dfb8 100644 --- a/frontend/src/app/workspace/component/result-panel/result-table-frame/result-table-frame.component.ts +++ b/frontend/src/app/workspace/component/result-panel/result-table-frame/result-table-frame.component.ts @@ -142,6 +142,11 @@ export class ResultTableFrameComponent implements OnInit, OnChanges { ngOnChanges(changes: SimpleChanges): void { if (changes.operatorId) this.operatorId = changes.operatorId.currentValue; + // another port's rows are not this port's, so a new port starts from an empty table + if (changes.portIndex && !changes.portIndex.firstChange) { + this.currentResult = []; + this.currentColumns = undefined; + } if (this.operatorId) { const paginatedResultService = this.paginatedResultService(); if (paginatedResultService) { @@ -415,14 +420,24 @@ export class ResultTableFrameComponent implements OnInit, OnChanges { return; } this.isLoadingResult = true; + // A reply can arrive after the table moved on to another operator or port; it is not theirs. + const operatorId = this.operatorId; + const portIndex = this.portIndex; paginatedResultService .selectPage(this.currentPageIndex, this.pageSize, this.currentColumnOffset, this.columnLimit, this.columnSearch) .pipe(untilDestroyed(this)) .subscribe(pageData => { - if (this.currentPageIndex === pageData.pageIndex) { - this.setupResultTable(pageData.table, paginatedResultService.getCurrentTotalNumTuples()); - this.changeDetectorRef.detectChanges(); + if ( + this.operatorId !== operatorId || + this.portIndex !== portIndex || + this.currentPageIndex !== pageData.pageIndex + ) { + return; } + // an empty page keeps the rows shown, but the loading is over + if (pageData.table.length === 0) this.isLoadingResult = false; + this.setupResultTable(pageData.table, paginatedResultService.getCurrentTotalNumTuples()); + this.changeDetectorRef.detectChanges(); }); } From a64630b3afcfcfe57f8355445cb4949be2ab8346 Mon Sep 17 00:00:00 2001 From: kary zheng Date: Tue, 20 Oct 2026 16:14:07 -0700 Subject: [PATCH 4/8] feat(frontend): preview an operator's result when hovering it on the canvas Resting the pointer on an operator for half a second shows a card beside it with the operator's first rows and its size, or its error, or how to keep a result the run did not keep. A pointer passing over, dragging, or a click opens nothing, and the card lingers a moment so the pointer can move into it. An operator with several outputs says the rows are the first output's. The preview can be turned off under Layers, and stays off while the heat-map overlay is on. The first rows come from a request of their own, so a preview never moves the result table's page or reads its cache. Co-Authored-By: Claude Opus 5.5 --- .../component/menu/menu.component.html | 8 + .../component/menu/menu.component.ts | 6 + .../operator-result-preview.component.html | 69 +++++ .../operator-result-preview.component.scss | 108 ++++++++ .../operator-result-preview.component.spec.ts | 247 ++++++++++++++++++ .../operator-result-preview.component.ts | 170 ++++++++++++ .../workflow-editor.component.html | 8 + .../workflow-editor.component.scss | 5 + .../workflow-editor.component.spec.ts | 35 ++- .../workflow-editor.component.ts | 89 +++++++ .../model/joint-graph-wrapper.ts | 15 ++ .../workflow-result.service.spec.ts | 47 +++- .../workflow-result.service.ts | 29 ++ 13 files changed, 833 insertions(+), 3 deletions(-) create mode 100644 frontend/src/app/workspace/component/workflow-editor/operator-result-preview/operator-result-preview.component.html create mode 100644 frontend/src/app/workspace/component/workflow-editor/operator-result-preview/operator-result-preview.component.scss create mode 100644 frontend/src/app/workspace/component/workflow-editor/operator-result-preview/operator-result-preview.component.spec.ts create mode 100644 frontend/src/app/workspace/component/workflow-editor/operator-result-preview/operator-result-preview.component.ts diff --git a/frontend/src/app/workspace/component/menu/menu.component.html b/frontend/src/app/workspace/component/menu/menu.component.html index c5bde9d41dc..c238fddd0ba 100644 --- a/frontend/src/app/workspace/component/menu/menu.component.html +++ b/frontend/src/app/workspace/component/menu/menu.component.html @@ -227,6 +227,14 @@ >Status +
  • + +