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..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 @@ -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,83 @@ 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("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", () => { @@ -344,6 +423,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 +670,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..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 @@ -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,14 @@ export class ResultTableFrameComponent implements OnInit, OnChanges { ) {} ngOnChanges(changes: SimpleChanges): void { - this.operatorId = changes.operatorId?.currentValue; + 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.workflowResultService.getPaginatedResultService(this.operatorId); + const paginatedResultService = this.paginatedResultService(); if (paginatedResultService) { this.isFrontPagination = false; this.totalNumTuples = paginatedResultService.getCurrentTotalNumTuples(); @@ -150,6 +161,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 +186,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 +208,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 +366,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,19 +415,29 @@ export class ResultTableFrameComponent implements OnInit, OnChanges { if (!this.operatorId) { return; } - const paginatedResultService = this.workflowResultService.getPaginatedResultService(this.operatorId); + const paginatedResultService = this.paginatedResultService(); if (!paginatedResultService) { 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(); }); } 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<{