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 30c845ed0fb..5a332de2c2d 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). @@ -513,7 +513,12 @@ class ExecutionResultService( // A page with no rows has no tuple to read the columns off, but the // stored result still declares them. .getOrElse(columns.get.map(schemaOption.get.getAttribute)) - PaginatedResultEvent.apply(request, mappedResults, attributes) + PaginatedResultEvent.apply( + request, + mappedResults, + attributes, + virtualDocument.getCount.toInt + ) case None => // Handle the case when storageUri is empty @@ -526,6 +531,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 5e4975d17f8..f911bf60799 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 @@ -937,6 +937,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() @@ -1249,6 +1275,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 => @@ -1462,10 +1516,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 ) @@ -1473,13 +1530,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 ) ) @@ -1504,9 +1562,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/app.module.ts b/frontend/src/app/app.module.ts index 3a0c6e6413f..10e7652efc7 100644 --- a/frontend/src/app/app.module.ts +++ b/frontend/src/app/app.module.ts @@ -133,7 +133,6 @@ import { FormlyNgZorroAntdModule } from "@ngx-formly/ng-zorro-antd"; import { FlarumComponent } from "./dashboard/component/user/flarum/flarum.component"; import { NzAlertModule } from "ng-zorro-antd/alert"; import { LeftPanelComponent } from "./workspace/component/left-panel/left-panel.component"; -import { ErrorFrameComponent } from "./workspace/component/result-panel/error-frame/error-frame.component"; import { NzResizableModule } from "ng-zorro-antd/resizable"; import { WorkflowRuntimeStatisticsComponent } from "./dashboard/component/user/user-workflow/ngbd-modal-workflow-executions/workflow-runtime-statistics/workflow-runtime-statistics.component"; import { TimeTravelComponent } from "./workspace/component/left-panel/time-travel/time-travel.component"; @@ -303,7 +302,6 @@ registerLocaleData(en); ShareAccessComponent, WorkflowExecutionHistoryComponent, ConsoleFrameComponent, - ErrorFrameComponent, ResultTableFrameComponent, OperatorPropertyEditFrameComponent, FilesUploaderComponent, 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 +
  • + +
    • +
  • + + pin result +
    • +
  • { expect(operatorMenuService.executeUpToOperator).toHaveBeenCalledTimes(1); }); + it("pin result emits pinResult with the highlighted operator", () => { + highlightedOperatorsSubject.next(["op1"]); + fixture.detectChanges(); + const pinned: string[] = []; + component.pinResult.subscribe(operatorId => pinned.push(operatorId)); + + clickItem("pin result"); + + expect(pinned).toEqual(["op1"]); + }); + + it("offers pin result for exactly one operator only", () => { + const labels = () => + fixture.debugElement.queryAll(By.css("li[nz-menu-item]")).map(li => norm(li.nativeElement.textContent)); + + highlightedOperatorsSubject.next(["op1", "op2"]); + fixture.detectChanges(); + expect(labels()).not.toContain("pin result"); + + highlightedOperatorsSubject.next(["op1"]); + highlightedCommentBoxesSubject.next(["comment1"]); + fixture.detectChanges(); + expect(labels()).not.toContain("pin result"); + }); + it("Export result invokes onClickExportHighlightedExecutionResult", () => { ( workflowResultExportService as unknown as { hasResultToExportOnHighlightedOperators: boolean } diff --git a/frontend/src/app/workspace/component/workflow-editor/context-menu/context-menu/context-menu.component.ts b/frontend/src/app/workspace/component/workflow-editor/context-menu/context-menu/context-menu.component.ts index f7a0888e00c..45aa9bc6d4f 100644 --- a/frontend/src/app/workspace/component/workflow-editor/context-menu/context-menu/context-menu.component.ts +++ b/frontend/src/app/workspace/component/workflow-editor/context-menu/context-menu/context-menu.component.ts @@ -17,7 +17,7 @@ * under the License. */ -import { Component, Input } from "@angular/core"; +import { Component, EventEmitter, Input, Output } from "@angular/core"; import { OperatorMenuService } from "src/app/workspace/service/operator-menu/operator-menu.service"; import { WorkflowActionService } from "src/app/workspace/service/workflow-graph/model/workflow-action.service"; import { UntilDestroy, untilDestroyed } from "@ngneat/until-destroy"; @@ -51,6 +51,12 @@ export class ContextMenuComponent { */ @Input() structureLocked = false; + /** + * Asks the editor that owns this menu to pin an operator's result window. The hover card's Pin + * button is the other way, and it is gone while Result preview is off or the heat-map is on. + */ + @Output() pinResult = new EventEmitter(); + /** Whether the graph may be re-shaped from this menu: modification on, and no structure lock. */ public get canModify(): boolean { return this.isWorkflowModifiable && !this.structureLocked; @@ -123,6 +129,10 @@ export class ContextMenuComponent { this.operatorMenuService.saveHighlightedElements(); } + public onPinResult(): void { + this.pinResult.emit(this.highlightedOperatorIds[0]); + } + public onPaste(): void { this.operatorMenuService.performPasteOperation(); } diff --git a/frontend/src/app/workspace/component/workflow-editor/operator-result-preview/operator-result-preview.component.html b/frontend/src/app/workspace/component/workflow-editor/operator-result-preview/operator-result-preview.component.html new file mode 100644 index 00000000000..46e574c1d1f --- /dev/null +++ b/frontend/src/app/workspace/component/workflow-editor/operator-result-preview/operator-result-preview.component.html @@ -0,0 +1,80 @@ + + + diff --git a/frontend/src/app/workspace/component/workflow-editor/operator-result-preview/operator-result-preview.component.scss b/frontend/src/app/workspace/component/workflow-editor/operator-result-preview/operator-result-preview.component.scss new file mode 100644 index 00000000000..04c99709320 --- /dev/null +++ b/frontend/src/app/workspace/component/workflow-editor/operator-result-preview/operator-result-preview.component.scss @@ -0,0 +1,126 @@ +/** + * 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. + */ + +// One size for every operator, so the card does not jump as the pointer moves between them: it +// fits the 5 rows and 6 columns the preview shows, and longer content is cut or scrolls inside. +.result-preview { + box-sizing: border-box; + width: 440px; + height: 210px; + overflow: hidden; + padding: 10px 12px; + display: flex; + flex-direction: column; + gap: 8px; + font-size: 12px; + line-height: 1.4; + color: #262626; + background: white; + border: 1px solid #d9d9d9; + border-radius: 6px; + box-shadow: 0 6px 16px rgba(0, 0, 0, 0.12); + + &__header { + flex-shrink: 0; + display: flex; + justify-content: space-between; + align-items: baseline; + gap: 8px; + } + + &__pin { + flex-shrink: 0; + display: flex; + align-items: center; + gap: 4px; + padding: 1px 8px; + font-size: 12px; + color: #1890ff; + background: white; + border: 1px solid #91d5ff; + border-radius: 4px; + cursor: pointer; + + &:hover { + background: #e6f7ff; + } + } + + &__title { + min-width: 0; + font-weight: 600; + overflow: hidden; + text-overflow: ellipsis; + white-space: nowrap; + } + + &__summary { + flex-shrink: 0; + color: #595959; + + &--error { + color: #cf1322; + } + } + + // the columns share the card's width evenly, whatever their contents + &__table { + flex-shrink: 0; + width: 100%; + table-layout: fixed; + border-collapse: collapse; + font-family: monospace; + font-size: 11px; + + th, + td { + padding: 3px 8px; + overflow: hidden; + text-overflow: ellipsis; + white-space: nowrap; + text-align: left; + border-bottom: 1px solid #f0f0f0; + } + + th { + font-weight: 600; + color: #595959; + background: #fafafa; + } + } + + &__note { + flex-shrink: 0; + color: #595959; + } + + // a long error message scrolls inside the card instead of growing it + &__detail { + min-height: 0; + overflow-y: auto; + padding: 6px 8px; + border-radius: 4px; + background: #f5f5f5; + + &--error { + color: #a8071a; + background: #fff1f0; + } + } +} diff --git a/frontend/src/app/workspace/component/workflow-editor/operator-result-preview/operator-result-preview.component.spec.ts b/frontend/src/app/workspace/component/workflow-editor/operator-result-preview/operator-result-preview.component.spec.ts new file mode 100644 index 00000000000..8e0e6b042a1 --- /dev/null +++ b/frontend/src/app/workspace/component/workflow-editor/operator-result-preview/operator-result-preview.component.spec.ts @@ -0,0 +1,247 @@ +/** + * 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 { ComponentFixture, TestBed } from "@angular/core/testing"; +import { HttpClientTestingModule } from "@angular/common/http/testing"; +import { By } from "@angular/platform-browser"; +import { NzModalModule } from "ng-zorro-antd/modal"; + +import { NEVER, of } from "rxjs"; +import { + formatPreviewCell, + OperatorResultPreviewComponent, + PREVIEW_CELL_LIMIT, + PREVIEW_COLUMN_LIMIT, + PREVIEW_ROW_LIMIT, +} from "./operator-result-preview.component"; +import { WorkflowActionService } from "../../../service/workflow-graph/model/workflow-action.service"; +import { WorkflowResultService } from "../../../service/workflow-result/workflow-result.service"; +import { WorkflowCompilingService } from "../../../service/compile-workflow/workflow-compiling.service"; +import { ExecuteWorkflowService } from "../../../service/execute-workflow/execute-workflow.service"; +import { OperatorMetadataService } from "../../../service/operator-metadata/operator-metadata.service"; +import { StubOperatorMetadataService } from "../../../service/operator-metadata/stub-operator-metadata.service"; +import { ComputingUnitStatusService } from "../../../../common/service/computing-unit/computing-unit-status/computing-unit-status.service"; +import { MockComputingUnitStatusService } from "../../../../common/service/computing-unit/computing-unit-status/mock-computing-unit-status.service"; +import { commonTestProviders } from "../../../../common/testing/test-utils"; +import { mockPoint, mockScanPredicate } from "../../../service/workflow-graph/model/mock-workflow-data"; +import { ExecutionState } from "../../../types/execute-workflow.interface"; +import { ConsoleMessage } from "../../../types/workflow-common.interface"; +import { WorkflowConsoleService } from "../../../service/workflow-console/workflow-console.service"; +import { AttributeType } from "../../../types/workflow-compiling.interface"; +import { WorkflowFatalError } from "../../../types/workflow-websocket.interface"; +import { OperatorPaginationResultService } from "../../../service/workflow-result/workflow-result.service"; + +describe("OperatorResultPreviewComponent", () => { + let component: OperatorResultPreviewComponent; + let fixture: ComponentFixture; + let resultService: WorkflowResultService; + let compilingService: WorkflowCompilingService; + let executeService: ExecuteWorkflowService; + + const id = mockScanPredicate.operatorID; + + const fatalError = (operatorId: string): WorkflowFatalError => ({ + message: "KeyError: 'price'", + details: "", + operatorId, + workerId: "", + type: { name: "ExecutionError" }, + timestamp: { nanos: 0, seconds: 0 }, + }); + + const withColumns = (count: number) => + vi.spyOn(compilingService, "getOperatorOutputSchemaMap").mockReturnValue({ + "output-0": Array.from({ length: count }, (_, i) => ({ + attributeName: `col${i}`, + attributeType: "string" as AttributeType, + })), + }); + + // A paginated result of `rows` tuples whose first page is `page`; `page` undefined never answers. + const withRows = (rows: number, page?: Record[]) => { + const peekFirstRows = vi.fn(() => (page ? of({ table: page }) : NEVER)); + vi.spyOn(resultService, "getPaginatedResultService").mockReturnValue({ + getCurrentTotalNumTuples: () => rows, + peekFirstRows, + } as unknown as OperatorPaginationResultService); + return peekFirstRows; + }; + + // Set through the input, as the canvas does, so ngOnChanges loads the rows. + const hover = (operatorId: string) => { + fixture.componentRef.setInput("operatorId", operatorId); + fixture.detectChanges(); + }; + + beforeEach(async () => { + await TestBed.configureTestingModule({ + imports: [OperatorResultPreviewComponent, HttpClientTestingModule, NzModalModule], + providers: [ + WorkflowActionService, + ExecuteWorkflowService, + { provide: OperatorMetadataService, useClass: StubOperatorMetadataService }, + { provide: ComputingUnitStatusService, useClass: MockComputingUnitStatusService }, + ...commonTestProviders, + ], + }).compileComponents(); + + fixture = TestBed.createComponent(OperatorResultPreviewComponent); + component = fixture.componentInstance; + resultService = TestBed.inject(WorkflowResultService); + compilingService = TestBed.inject(WorkflowCompilingService); + executeService = TestBed.inject(ExecuteWorkflowService); + TestBed.inject(WorkflowActionService).addOperator({ ...mockScanPredicate, customDisplayName: "Orders" }, mockPoint); + component.operatorId = id; + }); + + afterEach(() => { + vi.restoreAllMocks(); + fixture.destroy(); + }); + + it("shows a table result's size and its columns", () => { + withRows(1240); + withColumns(3); + + expect(component.preview).toEqual({ + title: "Orders", + kind: "table", + summary: "1,240 rows · 3 columns", + columns: ["col0", "col1", "col2"], + moreColumns: 0, + }); + }); + + it("lists only the first columns of a wide result and counts the rest", () => { + withRows(1); + withColumns(PREVIEW_COLUMN_LIMIT + 4); + + const preview = component.preview; + expect(preview.summary).toBe(`1 row · ${PREVIEW_COLUMN_LIMIT + 4} columns`); + expect(preview.columns.length).toBe(PREVIEW_COLUMN_LIMIT); + expect(preview.moreColumns).toBe(4); + }); + + it("says the rows are the first output's when the operator has several", () => { + const split = "split-op"; + TestBed.inject(WorkflowActionService).addOperator( + { + ...mockScanPredicate, + operatorID: split, + outputPorts: [{ portID: "output-0" }, { portID: "output-1" }], + }, + mockPoint + ); + component.operatorId = split; + withRows(10); + withColumns(2); + + expect(component.preview.summary).toBe("10 rows · 2 columns · first of 2 outputs"); + }); + + it("says a chart was drawn for a result that is not a table", () => { + vi.spyOn(resultService, "hasResult").mockReturnValue(true); + + expect(component.preview.kind).toBe("chart"); + }); + + it("shows the operator's error ahead of any result", () => { + withRows(5); + vi.spyOn(executeService, "getErrorMessages").mockReturnValue([fatalError(id)]); + + expect(component.preview).toMatchObject({ kind: "error", summary: "failed", detail: "KeyError: 'price'" }); + }); + + // A bad tuple pauses the worker and writes an ERROR to the console rather than failing the run. + it("shows the error an operator wrote to its console", () => { + withRows(5); + vi.spyOn(TestBed.inject(WorkflowConsoleService), "getConsoleMessages").mockReturnValue([ + { msgType: { name: "PRINT" }, title: "hello" }, + { msgType: { name: "ERROR" }, title: 'NumberFormatException: For input string: "a"' }, + ] as unknown as ConsoleMessage[]); + + expect(component.preview).toMatchObject({ + kind: "error", + summary: "failed", + detail: 'NumberFormatException: For input string: "a"', + }); + }); + + it("says to run the workflow when nothing has run yet", () => { + expect(component.preview).toMatchObject({ kind: "none", detail: "Run the workflow to see its result." }); + }); + + it("says how to keep the result when the run did not keep one", () => { + vi.spyOn(executeService, "getExecutionState").mockReturnValue({ state: ExecutionState.Completed }); + + expect(component.preview.kind).toBe("none"); + expect(component.preview.detail).toContain("Turn on its eye icon"); + }); + + it("renders the first rows under the first columns, and counts the columns left out", () => { + withRows(2, [ + { col0: "a", col1: 1, col6: "hidden" }, + { col0: null, col1: { nested: true }, col6: "hidden" }, + ]); + withColumns(PREVIEW_COLUMN_LIMIT + 1); + hover(id); + + const text = (css: string) => fixture.debugElement.query(By.css(css)).nativeElement.textContent.trim(); + const cells = (css: string) => + fixture.debugElement.queryAll(By.css(css)).map(e => e.nativeElement.textContent.trim()); + expect(text(".result-preview__title")).toBe("Orders"); + expect(text(".result-preview__summary")).toBe(`2 rows · ${PREVIEW_COLUMN_LIMIT + 1} columns`); + expect(cells("th").length).toBe(PREVIEW_COLUMN_LIMIT); + expect(fixture.debugElement.queryAll(By.css("tbody tr")).length).toBe(2); + expect(cells("tbody tr:first-child td").slice(0, 2)).toEqual(["a", "1"]); + expect(cells("tbody tr:nth-child(2) td").slice(0, 2)).toEqual(["", '{"nested":true}']); + expect(text(".result-preview__note")).toBe("+1 more column"); + expect(fixture.debugElement.query(By.css(".result-preview__detail"))).toBeNull(); + }); + + it("asks for the first rows only, and shows that they are loading until they come", () => { + const peek = withRows(10); + withColumns(2); + hover(id); + + expect(peek).toHaveBeenCalledWith(PREVIEW_ROW_LIMIT); + expect(component.rows).toBeUndefined(); + expect(fixture.debugElement.query(By.css(".result-preview__note")).nativeElement.textContent.trim()).toBe( + "Loading rows…" + ); + }); + + it("asks for no rows when the result has none", () => { + const peek = withRows(0); + withColumns(2); + hover(id); + + expect(peek).not.toHaveBeenCalled(); + expect(component.rows).toEqual([]); + }); + + it("prints a cell short, blank for null, and as JSON for an object", () => { + expect(formatPreviewCell(null)).toBe(""); + expect(formatPreviewCell(undefined)).toBe(""); + expect(formatPreviewCell(3.5)).toBe("3.5"); + expect(formatPreviewCell({ a: 1 })).toBe('{"a":1}'); + const long = "x".repeat(PREVIEW_CELL_LIMIT + 10); + expect(formatPreviewCell(long)).toBe("x".repeat(PREVIEW_CELL_LIMIT - 1) + "…"); + }); +}); diff --git a/frontend/src/app/workspace/component/workflow-editor/operator-result-preview/operator-result-preview.component.ts b/frontend/src/app/workspace/component/workflow-editor/operator-result-preview/operator-result-preview.component.ts new file mode 100644 index 00000000000..8369ff589d5 --- /dev/null +++ b/frontend/src/app/workspace/component/workflow-editor/operator-result-preview/operator-result-preview.component.ts @@ -0,0 +1,174 @@ +/** + * 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 { ChangeDetectorRef, Component, EventEmitter, Input, OnChanges, Output, SimpleChanges } from "@angular/core"; +import { NgFor, NgIf } from "@angular/common"; +import { NzIconDirective } from "ng-zorro-antd/icon"; +import { Subscription } from "rxjs"; +import { UntilDestroy, untilDestroyed } from "@ngneat/until-destroy"; +import { WorkflowActionService } from "../../../service/workflow-graph/model/workflow-action.service"; +import { WorkflowResultService } from "../../../service/workflow-result/workflow-result.service"; +import { WorkflowCompilingService } from "../../../service/compile-workflow/workflow-compiling.service"; +import { ExecuteWorkflowService } from "../../../service/execute-workflow/execute-workflow.service"; +import { WorkflowConsoleService } from "../../../service/workflow-console/workflow-console.service"; +import { ExecutionState } from "../../../types/execute-workflow.interface"; + +/** How many columns the preview shows before it says how many more there are. */ +export const PREVIEW_COLUMN_LIMIT = 6; +/** How many rows the preview shows. */ +export const PREVIEW_ROW_LIMIT = 5; +/** How many characters of a cell the preview shows; the rest is in the cell's tooltip. */ +export const PREVIEW_CELL_LIMIT = 24; + +/** A cell as the preview prints it: short, and never "[object Object]". */ +export function formatPreviewCell(value: unknown): string { + if (value === null || value === undefined) return ""; + const text = typeof value === "object" ? JSON.stringify(value) : String(value); + return text.length > PREVIEW_CELL_LIMIT ? text.slice(0, PREVIEW_CELL_LIMIT - 1) + "…" : text; +} + +export interface OperatorResultPreview { + title: string; + kind: "table" | "chart" | "error" | "none"; + summary: string; + columns: readonly string[]; + moreColumns: number; + detail?: string; +} + +/** + * What hovering an operator on the canvas shows about its result: the first rows of a table, + * that it drew a chart, its error, or why there is nothing. The rows are fetched on their own + * request, which leaves the page the result table is on alone. + */ +@UntilDestroy() +@Component({ + selector: "texera-operator-result-preview", + templateUrl: "./operator-result-preview.component.html", + styleUrls: ["./operator-result-preview.component.scss"], + imports: [NgIf, NgFor, NzIconDirective], +}) +export class OperatorResultPreviewComponent implements OnChanges { + @Input() operatorId = ""; + // whether the card offers to pin the result next to the operator + @Input() pinnable = false; + @Output() pin = new EventEmitter(); + + // The first rows of a table result, one string per shown column; undefined while they load. + rows: readonly (readonly string[])[] | undefined = undefined; + private rowsSubscription?: Subscription; + + constructor( + private workflowActionService: WorkflowActionService, + private workflowResultService: WorkflowResultService, + private workflowCompilingService: WorkflowCompilingService, + private executeWorkflowService: ExecuteWorkflowService, + private workflowConsoleService: WorkflowConsoleService, + private changeDetectorRef: ChangeDetectorRef + ) {} + + ngOnChanges(changes: SimpleChanges): void { + if (changes["operatorId"]) { + this.loadRows(); + } + } + + private loadRows(): void { + this.rowsSubscription?.unsubscribe(); + const paginated = this.workflowResultService.getPaginatedResultService(this.operatorId); + if (!paginated || paginated.getCurrentTotalNumTuples() === 0) { + this.rows = []; + return; + } + this.rows = undefined; + const columns = this.outputColumns(this.operatorId).slice(0, PREVIEW_COLUMN_LIMIT); + // Replaced when another operator is hovered, and dropped with the preview when the pointer leaves. + this.rowsSubscription = paginated + .peekFirstRows(PREVIEW_ROW_LIMIT) + .pipe(untilDestroyed(this)) + .subscribe(page => { + this.rows = page.table.map(row => + columns.map(column => formatPreviewCell((row as Record)[column])) + ); + this.changeDetectorRef.detectChanges(); + }); + } + + get preview(): OperatorResultPreview { + const id = this.operatorId; + const graph = this.workflowActionService.getTexeraGraph(); + const operator = graph.hasOperator(id) ? graph.getOperator(id) : undefined; + const title = operator?.customDisplayName ?? operator?.operatorType ?? id; + const base = { title, columns: [], moreColumns: 0 }; + + const error = this.executeWorkflowService + .getErrorMessages() + .concat(Object.values(this.workflowCompilingService.getWorkflowCompilationErrors())) + .find(e => e.operatorId === id); + if (error) { + return { ...base, kind: "error", summary: "failed", detail: error.message }; + } + // A bad tuple pauses the worker and writes an ERROR to the console rather than failing the run. + const consoleError = (this.workflowConsoleService.getConsoleMessages(id) ?? []).find( + message => message.msgType.name === "ERROR" + ); + if (consoleError) { + return { ...base, kind: "error", summary: "failed", detail: consoleError.title }; + } + + const paginated = this.workflowResultService.getPaginatedResultService(id); + if (paginated) { + const columns = this.outputColumns(id); + const rows = paginated.getCurrentTotalNumTuples(); + // The preview is of the first output port; the pinned window has a tab for each. + const outputCount = operator?.outputPorts.length ?? 1; + return { + ...base, + kind: "table", + summary: + `${rows.toLocaleString()} ${rows === 1 ? "row" : "rows"} · ${columns.length} ${ + columns.length === 1 ? "column" : "columns" + }` + (outputCount > 1 ? ` · first of ${outputCount} outputs` : ""), + columns: columns.slice(0, PREVIEW_COLUMN_LIMIT), + moreColumns: Math.max(0, columns.length - PREVIEW_COLUMN_LIMIT), + }; + } + + if (this.workflowResultService.hasResult(id)) { + return { ...base, kind: "chart", summary: "chart" }; + } + + const hasRun = this.executeWorkflowService.getExecutionState().state !== ExecutionState.Uninitialized; + return { + ...base, + kind: "none", + summary: "no result", + detail: hasRun + ? "This run did not keep this operator's result. Turn on its eye icon to keep it next time." + : "Run the workflow to see its result.", + }; + } + + /** The columns the operator's first output port was compiled to produce. */ + private outputColumns(operatorId: string): string[] { + const ports = this.workflowCompilingService.getOperatorOutputSchemaMap(operatorId); + const firstPort = ports ? Object.values(ports)[0] : undefined; + return (firstPort ?? []).map(attribute => attribute.attributeName); + } +} diff --git a/frontend/src/app/workspace/component/workflow-editor/pinned-result-window/pinned-result-window.component.html b/frontend/src/app/workspace/component/workflow-editor/pinned-result-window/pinned-result-window.component.html new file mode 100644 index 00000000000..142d045c893 --- /dev/null +++ b/frontend/src/app/workspace/component/workflow-editor/pinned-result-window/pinned-result-window.component.html @@ -0,0 +1,70 @@ + + +
    +
    + + {{ title }} + +
    +
    + +
    +
    + + + +
    +
    diff --git a/frontend/src/app/workspace/component/workflow-editor/pinned-result-window/pinned-result-window.component.scss b/frontend/src/app/workspace/component/workflow-editor/pinned-result-window/pinned-result-window.component.scss new file mode 100644 index 00000000000..6d25a26c344 --- /dev/null +++ b/frontend/src/app/workspace/component/workflow-editor/pinned-result-window/pinned-result-window.component.scss @@ -0,0 +1,102 @@ +/** + * 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. + */ + +.pinned-result { + width: 520px; + // tall enough for the table's first page to hold 5 rows, as the result panel's does + height: 480px; + min-width: 280px; + min-height: 160px; + display: flex; + flex-direction: column; + // the bottom-right corner resizes the window + resize: both; + overflow: hidden; + background: white; + border: 1px solid #d9d9d9; + border-radius: 6px; + box-shadow: 0 6px 16px rgba(0, 0, 0, 0.16); + + &__header { + flex-shrink: 0; + display: flex; + align-items: center; + gap: 6px; + height: 30px; + padding: 0 4px 0 10px; + cursor: move; + user-select: none; + background: #fafafa; + border-bottom: 1px solid #f0f0f0; + } + + &__pin { + color: #1890ff; + } + + &__title { + flex: 1; + min-width: 0; + font-weight: 600; + overflow: hidden; + text-overflow: ellipsis; + white-space: nowrap; + } + + &__close { + border: none; + background: transparent; + cursor: pointer; + color: #595959; + padding: 2px 6px; + } + + &__ports { + flex-shrink: 0; + display: flex; + gap: 2px; + overflow-x: auto; + padding: 0 6px; + border-bottom: 1px solid #f0f0f0; + } + + &__port { + flex-shrink: 0; + height: 28px; + padding: 0 10px; + border: none; + border-bottom: 2px solid transparent; + background: transparent; + cursor: pointer; + color: #595959; + + &--active { + border-bottom-color: #1890ff; + color: #1890ff; + font-weight: 600; + } + } + + &__body { + flex: 1; + min-height: 0; + overflow: auto; + padding: 4px; + } +} diff --git a/frontend/src/app/workspace/component/workflow-editor/pinned-result-window/pinned-result-window.component.spec.ts b/frontend/src/app/workspace/component/workflow-editor/pinned-result-window/pinned-result-window.component.spec.ts new file mode 100644 index 00000000000..4ea085e6766 --- /dev/null +++ b/frontend/src/app/workspace/component/workflow-editor/pinned-result-window/pinned-result-window.component.spec.ts @@ -0,0 +1,140 @@ +/** + * 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 { Component, Input } from "@angular/core"; +import { ComponentFixture, TestBed } from "@angular/core/testing"; +import { By } from "@angular/platform-browser"; +import { PinnedResultWindowComponent } from "./pinned-result-window.component"; +import { ResultTableFrameComponent } from "../../result-panel/result-table-frame/result-table-frame.component"; +import { VisualizationFrameContentComponent } from "../../visualization-panel-content/visualization-frame-content.component"; +import { OperatorResultPreviewComponent } from "../operator-result-preview/operator-result-preview.component"; +import { WorkflowActionService } from "../../../service/workflow-graph/model/workflow-action.service"; +import { + OperatorPaginationResultService, + WorkflowResultService, +} from "../../../service/workflow-result/workflow-result.service"; +import { OperatorPredicate, PortDescription } from "../../../types/workflow-common.interface"; + +// Stand-ins for the frames the window holds, so the test needs none of their services. +@Component({ selector: "texera-result-table-frame", template: "", standalone: true }) +class StubTableFrame { + @Input() operatorId?: string; + @Input() portIndex = 0; + @Input() containerHeight?: number; +} + +@Component({ selector: "texera-visualization-panel-content", template: "", standalone: true }) +class StubChartFrame { + @Input() operatorId?: string; +} + +@Component({ selector: "texera-operator-result-preview", template: "", standalone: true }) +class StubPreview { + @Input() operatorId?: string; +} + +describe("PinnedResultWindowComponent", () => { + let fixture: ComponentFixture; + let operator: OperatorPredicate; + let hasTable: boolean; + + const ports = (...names: (string | undefined)[]): PortDescription[] => + names.map((displayName, i) => ({ portID: `output-${i}`, displayName })); + + beforeEach(async () => { + operator = { operatorID: "op1", operatorType: "Split", outputPorts: ports(undefined) } as OperatorPredicate; + hasTable = true; + const graph = { hasOperator: () => true, getOperator: () => operator }; + + await TestBed.configureTestingModule({ + imports: [PinnedResultWindowComponent], + providers: [ + { provide: WorkflowActionService, useValue: { getTexeraGraph: () => graph } }, + { + provide: WorkflowResultService, + useValue: { + getPaginatedResultService: () => (hasTable ? ({} as OperatorPaginationResultService) : undefined), + getResultService: () => undefined, + }, + }, + ], + }) + .overrideComponent(PinnedResultWindowComponent, { + remove: { + imports: [ResultTableFrameComponent, VisualizationFrameContentComponent, OperatorResultPreviewComponent], + }, + add: { imports: [StubTableFrame, StubChartFrame, StubPreview] }, + }) + .compileComponents(); + + fixture = TestBed.createComponent(PinnedResultWindowComponent); + fixture.componentRef.setInput("operatorId", "op1"); + }); + + const tabs = () => fixture.debugElement.queryAll(By.css(".pinned-result__port")); + const table = () => fixture.debugElement.query(By.directive(StubTableFrame)).componentInstance as StubTableFrame; + + it("shows no port tabs for an operator with one output", () => { + fixture.detectChanges(); + + expect(tabs().length).toBe(0); + expect(table().portIndex).toBe(0); + }); + + it("shows a tab for each output, named as on the canvas or by its place, the first one shown", () => { + operator = { ...operator, outputPorts: ports("False", "True") }; + fixture.detectChanges(); + expect(tabs().map(tab => (tab.nativeElement as HTMLElement).textContent?.trim())).toEqual(["False", "True"]); + + operator = { ...operator, outputPorts: ports(undefined, "") }; + fixture.detectChanges(); + expect(tabs().map(tab => (tab.nativeElement as HTMLElement).textContent?.trim())).toEqual(["Output 1", "Output 2"]); + expect(tabs()[0].classes["pinned-result__port--active"]).toBe(true); + expect(table().portIndex).toBe(0); + }); + + it("shows the output whose tab is clicked", () => { + operator = { ...operator, outputPorts: ports("False", "True") }; + fixture.detectChanges(); + + (tabs()[1].nativeElement as HTMLElement).click(); + fixture.detectChanges(); + + expect(table().portIndex).toBe(1); + expect(table().operatorId).toBe("op1"); + expect(tabs()[1].attributes["aria-selected"]).toBe("true"); + expect(tabs()[0].attributes["aria-selected"]).toBe("false"); + }); + + it("hands the table the height its body leaves, not the result panel's", () => { + fixture.detectChanges(); + fixture.componentInstance.tableHeight = 400; + fixture.detectChanges(); + + expect(table().containerHeight).toBe(400); + }); + + it("shows no port tabs when the result is not a table", () => { + operator = { ...operator, outputPorts: ports("False", "True") }; + hasTable = false; + fixture.detectChanges(); + + expect(tabs().length).toBe(0); + }); +}); diff --git a/frontend/src/app/workspace/component/workflow-editor/pinned-result-window/pinned-result-window.component.ts b/frontend/src/app/workspace/component/workflow-editor/pinned-result-window/pinned-result-window.component.ts new file mode 100644 index 00000000000..7fb35446cfa --- /dev/null +++ b/frontend/src/app/workspace/component/workflow-editor/pinned-result-window/pinned-result-window.component.ts @@ -0,0 +1,134 @@ +/** + * 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 { + AfterViewInit, + ChangeDetectorRef, + Component, + ElementRef, + EventEmitter, + Input, + OnDestroy, + Output, + ViewChild, +} from "@angular/core"; +import { NgFor, NgIf } from "@angular/common"; +import { NzIconDirective } from "ng-zorro-antd/icon"; +import { WorkflowActionService } from "../../../service/workflow-graph/model/workflow-action.service"; +import { WorkflowResultService } from "../../../service/workflow-result/workflow-result.service"; +import { ResultTableFrameComponent } from "../../result-panel/result-table-frame/result-table-frame.component"; +import { VisualizationFrameContentComponent } from "../../visualization-panel-content/visualization-frame-content.component"; +import { OperatorResultPreviewComponent } from "../operator-result-preview/operator-result-preview.component"; + +/** + * An operator's result pinned next to it on the canvas. It shows the whole result, the way the + * result panel would, and stays until it is closed. Dragging its header moves it away from the + * operator; the editor keeps it at that offset when the operator or the canvas moves. + */ +@Component({ + selector: "texera-pinned-result-window", + templateUrl: "./pinned-result-window.component.html", + styleUrls: ["./pinned-result-window.component.scss"], + imports: [ + NgIf, + NgFor, + NzIconDirective, + ResultTableFrameComponent, + VisualizationFrameContentComponent, + OperatorResultPreviewComponent, + ], +}) +export class PinnedResultWindowComponent implements AfterViewInit, OnDestroy { + @Input() operatorId = ""; + @Output() closed = new EventEmitter(); + @Output() moved = new EventEmitter<{ dx: number; dy: number }>(); + @ViewChild("body") bodyElement?: ElementRef; + + /** The height the window's body leaves for the table, which fills it. */ + tableHeight?: number; + private resizeObserver?: ResizeObserver; + + constructor( + private workflowActionService: WorkflowActionService, + private workflowResultService: WorkflowResultService, + private changeDetectorRef: ChangeDetectorRef + ) {} + + ngAfterViewInit(): void { + const element = this.bodyElement?.nativeElement; + if (!element || typeof ResizeObserver === "undefined") return; + // the user resizes the window by its corner, so the body's height is read as it changes + this.resizeObserver = new ResizeObserver(() => { + const style = getComputedStyle(element); + const height = element.clientHeight - parseFloat(style.paddingTop) - parseFloat(style.paddingBottom); + if (height === this.tableHeight) return; + this.tableHeight = height; + this.changeDetectorRef.detectChanges(); + }); + this.resizeObserver.observe(element); + } + + ngOnDestroy(): void { + this.resizeObserver?.disconnect(); + } + + get title(): string { + const graph = this.workflowActionService.getTexeraGraph(); + if (!graph.hasOperator(this.operatorId)) return this.operatorId; + const operator = graph.getOperator(this.operatorId); + return operator.customDisplayName ?? operator.operatorType; + } + + /** The output port whose result is shown. */ + selectedPort = 0; + + /** The operator's output ports, each named as on the canvas, or by its place when unnamed. */ + get outputPorts(): { index: number; label: string }[] { + const graph = this.workflowActionService.getTexeraGraph(); + if (!graph.hasOperator(this.operatorId)) return []; + return graph + .getOperator(this.operatorId) + .outputPorts.map((port, index) => ({ index, label: port.displayName || `Output ${index + 1}` })); + } + + get kind(): "table" | "chart" | "other" { + if (this.workflowResultService.getPaginatedResultService(this.operatorId)) return "table"; + if (this.workflowResultService.getResultService(this.operatorId)) return "chart"; + return "other"; + } + + /** Drag by the header. Reports how far the pointer moved, so the editor can keep the offset. */ + startDrag(event: MouseEvent): void { + if ((event.target as HTMLElement).closest("button")) return; + event.preventDefault(); + let lastX = event.clientX; + let lastY = event.clientY; + const move = (e: MouseEvent) => { + this.moved.emit({ dx: e.clientX - lastX, dy: e.clientY - lastY }); + lastX = e.clientX; + lastY = e.clientY; + }; + const stop = () => { + window.removeEventListener("mousemove", move); + window.removeEventListener("mouseup", stop); + }; + window.addEventListener("mousemove", move); + window.addEventListener("mouseup", stop); + } +} diff --git a/frontend/src/app/workspace/component/workflow-editor/workflow-editor.component.html b/frontend/src/app/workspace/component/workflow-editor/workflow-editor.component.html index 43d25120e9c..7a03df1ccff 100644 --- a/frontend/src/app/workspace/component/workflow-editor/workflow-editor.component.html +++ b/frontend/src/app/workspace/component/workflow-editor/workflow-editor.component.html @@ -31,10 +31,39 @@
    {{ heatmapTooltip.metricLabel }}
    Heat: {{ heatmapTooltip.heatLabel }}
    + + + + + - + diff --git a/frontend/src/app/workspace/component/workflow-editor/workflow-editor.component.scss b/frontend/src/app/workspace/component/workflow-editor/workflow-editor.component.scss index ad7ce546fda..9074d8e0360 100644 --- a/frontend/src/app/workspace/component/workflow-editor/workflow-editor.component.scss +++ b/frontend/src/app/workspace/component/workflow-editor/workflow-editor.component.scss @@ -34,6 +34,35 @@ transition: fill 0.25s ease; } +.result-preview-anchor { + position: absolute; + z-index: 11; +} + +.pinned-result-anchor { + position: absolute; + z-index: 10; +} + +// One layer under the windows for the lines that tie each window to its operator. +.pinned-result-connectors { + position: absolute; + top: 0; + left: 0; + width: 100%; + height: 100%; + z-index: 9; + overflow: visible; + pointer-events: none; +} + +.pinned-result-connector { + fill: none; + stroke: #1890ff; + stroke-width: 1.5; + stroke-dasharray: 5 4; +} + .heatmap-tooltip { position: absolute; z-index: 11; diff --git a/frontend/src/app/workspace/component/workflow-editor/workflow-editor.component.spec.ts b/frontend/src/app/workspace/component/workflow-editor/workflow-editor.component.spec.ts index 3b419e683a3..1b355262cdb 100644 --- a/frontend/src/app/workspace/component/workflow-editor/workflow-editor.component.spec.ts +++ b/frontend/src/app/workspace/component/workflow-editor/workflow-editor.component.spec.ts @@ -24,7 +24,7 @@ import { DragDropService } from "../../service/drag-drop/drag-drop.service"; import { WorkflowUtilService } from "../../service/workflow-graph/util/workflow-util.service"; import { ComponentFixture, fakeAsync, TestBed, tick } from "@angular/core/testing"; import { ValidationWorkflowService } from "../../service/validation/validation-workflow.service"; -import { WorkflowEditorComponent } from "./workflow-editor.component"; +import { RESULT_PREVIEW_DELAY_MS, RESULT_PREVIEW_HIDE_MS, WorkflowEditorComponent } from "./workflow-editor.component"; import { workflowEditorTestImports, workflowEditorTestProviders } from "./workflow-editor.test-utils"; import { OperatorMetadataService } from "../../service/operator-metadata/operator-metadata.service"; import { StubOperatorMetadataService } from "../../service/operator-metadata/stub-operator-metadata.service"; @@ -374,6 +374,87 @@ describe("WorkflowEditorComponent", () => { expect(linkView.hasTools()).toBe(false); }); + it("previews a hovered operator's result once the pointer rests, and hides it when the pointer leaves", fakeAsync(() => { + addLinkedPair(); + const view = component.paper.findViewByModel(mockScanPredicate.operatorID); + + component.paper.trigger("element:mouseenter", view, new MouseEvent("mouseenter")); + // passing over an operator opens nothing; only a pointer that rests on it does + tick(RESULT_PREVIEW_DELAY_MS - 1); + expect(component.resultPreview).toBeNull(); + tick(1); + fixture.detectChanges(); + expect(component.resultPreview?.operatorId).toBe(mockScanPredicate.operatorID); + expect(fixture.debugElement.query(By.css("texera-operator-result-preview"))).toBeTruthy(); + + component.paper.trigger("element:mouseleave", view, new MouseEvent("mouseleave")); + // the card lingers a moment, so the pointer can cross into it + tick(RESULT_PREVIEW_HIDE_MS); + fixture.detectChanges(); + expect(component.resultPreview).toBeNull(); + expect(fixture.debugElement.query(By.css("texera-operator-result-preview"))).toBeNull(); + })); + + it("leaves the hover to the heat-map tooltip while that overlay is on", () => { + addLinkedPair(); + vi.spyOn(TestBed.inject(WorkflowActionService).getJointGraphWrapper(), "getHeatmapView").mockReturnValue( + "throughput" as any + ); + const view = component.paper.findViewByModel(mockScanPredicate.operatorID); + + component.paper.trigger("element:mouseenter", view, new MouseEvent("mouseenter")); + + expect(component.resultPreview).toBeNull(); + }); + + /** Right-click with the scan operator selected and pick "pin result" from the menu. */ + function pinScanFromContextMenu(): void { + TestBed.inject(WorkflowActionService).getJointGraphWrapper().highlightOperators(mockScanPredicate.operatorID); + fixture.detectChanges(); + const menu = fixture.debugElement.query(By.directive(NzDropdownMenuComponent)).componentInstance; + component.nzContextMenu.create(new MouseEvent("contextmenu", { clientX: 5, clientY: 5 }), menu); + fixture.detectChanges(); + const item = fixture.debugElement + .queryAll(By.css("texera-context-menu li[nz-menu-item]")) + .find(li => li.nativeElement.textContent.trim() === "pin result"); + expect(item).toBeTruthy(); + item!.triggerEventHandler("click", null); + fixture.detectChanges(); + component.nzContextMenu.close(); + } + + // With the hover card off there is no Pin button, so the right-click menu is the way to a result window. + it("pins an operator's result from the right-click menu while Result preview is off", () => { + const wrapper = TestBed.inject(WorkflowActionService).getJointGraphWrapper(); + wrapper.setResultPreviewEnabled(false); + try { + addLinkedPair(); + + pinScanFromContextMenu(); + pinScanFromContextMenu(); + + // pinning the same operator twice keeps one window + expect(component.pinnedResults.map(pinned => pinned.operatorId)).toEqual([mockScanPredicate.operatorID]); + const windows = fixture.debugElement.queryAll(By.css("texera-pinned-result-window")); + expect(windows.length).toBe(1); + expect(windows[0].nativeElement.getAttribute("data-operator-id")).toBe(mockScanPredicate.operatorID); + } finally { + wrapper.setResultPreviewEnabled(true); + } + }); + + it("pins an operator's result from the right-click menu while the heat-map is on", () => { + addLinkedPair(); + vi.spyOn(TestBed.inject(WorkflowActionService).getJointGraphWrapper(), "getHeatmapView").mockReturnValue( + "throughput" as any + ); + + pinScanFromContextMenu(); + + expect(component.pinnedResults.map(pinned => pinned.operatorId)).toEqual([mockScanPredicate.operatorID]); + expect(fixture.debugElement.query(By.css("texera-pinned-result-window"))).toBeTruthy(); + }); + /** Select the scan operator the way a click does and return its model, whose attrs say what is on show. */ function selectScanOperator(): joint.dia.Cell { addLinkedPair(); diff --git a/frontend/src/app/workspace/component/workflow-editor/workflow-editor.component.ts b/frontend/src/app/workspace/component/workflow-editor/workflow-editor.component.ts index 1fac6beeba3..a938227b301 100644 --- a/frontend/src/app/workspace/component/workflow-editor/workflow-editor.component.ts +++ b/frontend/src/app/workspace/component/workflow-editor/workflow-editor.component.ts @@ -52,9 +52,11 @@ import concaveman from "concaveman"; import { OperatorResultSummary, AgentService } from "../../service/agent/agent.service"; import { NzNoAnimationDirective } from "ng-zorro-antd/core/animation"; import { ContextMenuComponent } from "./context-menu/context-menu/context-menu.component"; -import { NgIf } from "@angular/common"; +import { NgFor, NgIf } from "@angular/common"; import { AgentInteractionComponent } from "../agent/agent-interaction/agent-interaction.component"; import { HeatmapLegendComponent } from "../heatmap-legend/heatmap-legend.component"; +import { OperatorResultPreviewComponent } from "./operator-result-preview/operator-result-preview.component"; +import { PinnedResultWindowComponent } from "./pinned-result-window/pinned-result-window.component"; import { JupyterPanelService } from "../../service/jupyter-panel/jupyter-panel.service"; // jointjs interactive options for enabling and disabling interactivity @@ -78,6 +80,19 @@ export const MAIN_CANVAS = { yMax: 1512, // yMin * 2.8 }; +/** How long the pointer rests on an operator before its result card opens. */ +export const RESULT_PREVIEW_DELAY_MS = 500; +/** How long the card stays after the pointer leaves, so the pointer can move into it. */ +export const RESULT_PREVIEW_HIDE_MS = 200; +/** The height of a pinned window's header, where the line from its operator comes in. */ +const PINNED_HEADER_HEIGHT = 30; + +/** Whether a mouse button is down. JointJS hands over a jQuery event, which keeps the DOM one inside. */ +function isButtonHeld(evt: unknown): boolean { + const event = evt as { buttons?: number; originalEvent?: { buttons?: number } }; + return (event.originalEvent?.buttons ?? event.buttons ?? 0) !== 0; +} + /** * WorkflowEditorComponent is the component for the main workflow editor part of the UI. * @@ -103,6 +118,9 @@ export const MAIN_CANVAS = { NgIf, AgentInteractionComponent, HeatmapLegendComponent, + OperatorResultPreviewComponent, + PinnedResultWindowComponent, + NgFor, ], }) export class WorkflowEditorComponent implements OnInit, AfterViewInit, OnDestroy { @@ -119,6 +137,12 @@ export class WorkflowEditorComponent implements OnInit, AfterViewInit, OnDestroy metricLabel: string; heatLabel: string; } | null = null; + // The hovered operator's result preview, and where to show it. Null when hidden. + public resultPreview: { operatorId: string; x: number; y: number } | null = null; + // Results pinned beside their operators, each with how far it was dragged from its spot. + public pinnedResults: { operatorId: string; dx: number; dy: number }[] = []; + private previewShowTimer?: ReturnType; + private previewHideTimer?: ReturnType; private paperInteractive: boolean = true; // Keeps the paper sized to its OWN container (not just the window) and rebuilds cell geometry // when the container goes 0 -> real size. Needed by embedded previews like the Form View strip, @@ -249,6 +273,7 @@ export class WorkflowEditorComponent implements OnInit, AfterViewInit, OnDestroy this.handleOperatorStatusUpdate(); this.handleHeatmapOverlay(); this.handleHeatmapHover(); + this.handleResultPreviewHover(); this.handleRegionEvents(); this.handleOperatorSuggestionHighlightEvent(); this.handleAgentHoverHighlight(); @@ -590,6 +615,178 @@ export class WorkflowEditorComponent implements OnInit, AfterViewInit, OnDestroy }); } + /** + * Shows the hovered operator's result preview beside it; its Pin button keeps the whole result + * there. Off when Layers > Result preview is unchecked, and while the heat-map overlay is on, + * whose own tooltip answers the hover then. + */ + private handleResultPreviewHover(): void { + fromJointPaperEvent(this.paper, "element:mouseenter") + .pipe(untilDestroyed(this)) + .subscribe(([elementView, evt]) => { + clearTimeout(this.previewHideTimer); + clearTimeout(this.previewShowTimer); + if (this.wrapper.getHeatmapView() !== null || !this.wrapper.isResultPreviewEnabled()) { + return; + } + const operatorId = elementView.model.id.toString(); + // no preview while a button is held: the pointer is dragging an operator or a link + if ( + !this.workflowActionService.getTexeraGraph().hasOperator(operatorId) || + isButtonHeld(evt) || + this.pinnedResults.some(pinned => pinned.operatorId === operatorId) + ) { + return; + } + // Only a pointer that rests on the operator opens the card, so passing over one does not. + this.previewShowTimer = setTimeout(() => { + this.resultPreview = { operatorId, ...this.operatorAnchor(operatorId) }; + // JointJS paper events fire outside Angular's zone (see handleHeatmapHover). + this.changeDetectorRef.detectChanges(); + }, RESULT_PREVIEW_DELAY_MS); + }); + + fromJointPaperEvent(this.paper, "element:mouseleave") + .pipe(untilDestroyed(this)) + .subscribe(() => { + clearTimeout(this.previewShowTimer); + this.hideResultPreviewSoon(); + }); + + // A click or a drag on the canvas closes the card at once. + merge(fromJointPaperEvent(this.paper, "element:pointerdown"), fromJointPaperEvent(this.paper, "blank:pointerdown")) + .pipe(untilDestroyed(this)) + .subscribe(() => { + clearTimeout(this.previewShowTimer); + if (this.resultPreview === null) return; + this.resultPreview = null; + this.changeDetectorRef.detectChanges(); + }); + + // Pinned windows sit beside their operators, so they move when an operator or the canvas does. + merge( + fromJointPaperEvent(this.paper, "translate"), + fromJointPaperEvent(this.paper, "scale"), + fromEvent(this.paper.model as any, "change:position") + ) + .pipe(untilDestroyed(this)) + .subscribe(() => { + if (this.pinnedResults.length > 0) this.changeDetectorRef.detectChanges(); + }); + + this.workflowActionService + .getTexeraGraph() + .getOperatorDeleteStream() + .pipe(untilDestroyed(this)) + .subscribe(({ deletedOperatorID }) => this.unpinResult(deletedOperatorID)); + } + + /** Close the card after a moment, so the pointer can cross from the operator into it. */ + public hideResultPreviewSoon(): void { + clearTimeout(this.previewHideTimer); + this.previewHideTimer = setTimeout(() => { + this.resultPreview = null; + this.changeDetectorRef.detectChanges(); + }, RESULT_PREVIEW_HIDE_MS); + } + + /** The pointer reached the card, so it stays open. */ + public keepResultPreview(): void { + clearTimeout(this.previewHideTimer); + } + + public pinResultPreview(): void { + if (this.resultPreview === null) return; + this.pinResult(this.resultPreview.operatorId); + } + + /** Pins an operator's result window. The right-click menu calls this too, so it works with the hover card off. */ + public pinResult(operatorId: string): void { + if (!this.pinnedResults.some(pinned => pinned.operatorId === operatorId)) { + this.pinnedResults = [...this.pinnedResults, { operatorId, dx: 0, dy: 0 }]; + } + this.resultPreview = null; + } + + public unpinResult(operatorId: string): void { + this.pinnedResults = this.pinnedResults.filter(pinned => pinned.operatorId !== operatorId); + this.changeDetectorRef.detectChanges(); + } + + // A window is its operator's, so a move keeps the window being dragged instead of rebuilding it. + public trackPinnedResult(_index: number, pinned: { operatorId: string }): string { + return pinned.operatorId; + } + + public movePinnedResult(operatorId: string, delta: { dx: number; dy: number }): void { + this.pinnedResults = this.pinnedResults.map(pinned => + pinned.operatorId === operatorId ? { ...pinned, dx: pinned.dx + delta.dx, dy: pinned.dy + delta.dy } : pinned + ); + } + + /** Where a pinned window goes: beside its operator, plus however far it was dragged from there. */ + public pinnedResultPosition(pinned: { operatorId: string; dx: number; dy: number }): { x: number; y: number } { + const anchor = this.operatorAnchor(pinned.operatorId); + return { x: anchor.x + pinned.dx, y: anchor.y + pinned.dy }; + } + + /** + * The line from an operator to its pinned window, a curve like the links between operators. + * Beside the operator, it runs side to side into the window's header; above or below, it runs + * from the operator's top or bottom to the window's nearest edge. + */ + public pinnedConnectorPath(pinned: { operatorId: string; dx: number; dy: number }): string { + const cell = this.paper.getModelById(pinned.operatorId); + if (!cell) return ""; + const bbox = cell.getBBox(); + const scale = this.paper.scale(); + const translate = this.paper.translate(); + const op = { + left: bbox.x * scale.sx + translate.tx, + right: (bbox.x + bbox.width) * scale.sx + translate.tx, + top: bbox.y * scale.sy + translate.ty, + bottom: (bbox.y + bbox.height) * scale.sy + translate.ty, + }; + const midX = (op.left + op.right) / 2; + const midY = (op.top + op.bottom) / 2; + + const corner = this.pinnedResultPosition(pinned); + const size = this.pinnedWindowSize(pinned.operatorId); + const win = { left: corner.x, right: corner.x + size.width, top: corner.y, bottom: corner.y + size.height }; + const headerY = win.top + PINNED_HEADER_HEIGHT / 2; + + const curve = (x1: number, y1: number, x2: number, y2: number, horizontal: boolean) => { + const bend = Math.max(30, (horizontal ? Math.abs(x2 - x1) : Math.abs(y2 - y1)) / 2); + const [bx, by] = horizontal ? [Math.sign(x2 - x1) * bend, 0] : [0, Math.sign(y2 - y1) * bend]; + return `M ${x1} ${y1} C ${x1 + bx} ${y1 + by}, ${x2 - bx} ${y2 - by}, ${x2} ${y2}`; + }; + if (win.left >= op.right) return curve(op.right, midY, win.left, headerY, true); + if (win.right <= op.left) return curve(op.left, midY, win.right, headerY, true); + // the window overlaps the operator's column: join the facing edges, at the point under the operator + const x = Math.min(Math.max(midX, win.left + 12), win.right - 12); + return win.top >= op.bottom ? curve(midX, op.bottom, x, win.top, false) : curve(midX, op.top, x, win.bottom, false); + } + + /** A pinned window's size as drawn: the user can resize it, so it is read from the page. */ + private pinnedWindowSize(operatorId: string): { width: number; height: number } { + const host = this.elementRef.nativeElement as HTMLElement; + const box = host.querySelector(`[data-operator-id="${operatorId}"] .pinned-result`); + return { width: box?.offsetWidth ?? 0, height: box?.offsetHeight ?? 0 }; + } + + /** Just right of the operator's box, level with its top, in the editor's coordinates. */ + private operatorAnchor(operatorId: string): { x: number; y: number } { + const cell = this.paper.getModelById(operatorId); + if (!cell) return { x: 0, y: 0 }; + const bbox = cell.getBBox(); + const scale = this.paper.scale(); + const translate = this.paper.translate(); + return { + x: (bbox.x + bbox.width) * scale.sx + translate.tx + 12, + y: bbox.y * scale.sy + translate.ty, + }; + } + /** * Resolves the execution state to render for an operator: an operator the * state map does not know falls back to Uninitialized, and a reported state diff --git a/frontend/src/app/workspace/service/workflow-console/workflow-console.service.spec.ts b/frontend/src/app/workspace/service/workflow-console/workflow-console.service.spec.ts index a098d805152..e99d643656e 100644 --- a/frontend/src/app/workspace/service/workflow-console/workflow-console.service.spec.ts +++ b/frontend/src/app/workspace/service/workflow-console/workflow-console.service.spec.ts @@ -80,6 +80,23 @@ describe("WorkflowConsoleService", () => { expect(updates).toBe(1); }); + it("lists every operator's messages together, oldest first, each with its operator", () => { + const at = (seconds: number, nanos: number, title: string): ConsoleMessage => ({ + ...consoleMessage("PRINT"), + timestamp: { seconds, nanos }, + title, + }); + consoleUpdateEvent$.next({ operatorId: "op1", messages: [at(1, 0, "a1"), at(3, 0, "a3")] }); + consoleUpdateEvent$.next({ operatorId: "op2", messages: [at(2, 0, "b2"), at(1, 500, "b1.5")] }); + + expect(service.getAllConsoleMessages().map(row => [row.operatorId, row.message.title])).toEqual([ + ["op1", "a1"], + ["op2", "b1.5"], + ["op2", "b2"], + ["op1", "a3"], + ]); + }); + it("clearConsoleMessages() removes all messages and notifies subscribers", () => { consoleUpdateEvent$.next({ operatorId: "op1", messages: [consoleMessage("PRINT")] }); expect(service.hasConsoleMessages("op1")).toBe(true); diff --git a/frontend/src/app/workspace/service/workflow-console/workflow-console.service.ts b/frontend/src/app/workspace/service/workflow-console/workflow-console.service.ts index 5f88a22ba9a..506d0c6a515 100644 --- a/frontend/src/app/workspace/service/workflow-console/workflow-console.service.ts +++ b/frontend/src/app/workspace/service/workflow-console/workflow-console.service.ts @@ -25,6 +25,8 @@ import { RingBuffer } from "ring-buffer-ts"; import { ExecutionState } from "../../types/execute-workflow.interface"; import { GuiConfigService } from "../../../common/service/gui-config.service"; +export type OperatorConsoleMessage = Readonly<{ operatorId: string; message: ConsoleMessage }>; + @Injectable({ providedIn: "root", }) @@ -66,6 +68,19 @@ export class WorkflowConsoleService { return this.consoleMessages.get(operatorId)?.toArray(); } + /** Every operator's messages in one list, oldest first, each with the operator it came from. */ + getAllConsoleMessages(): ReadonlyArray { + const all: OperatorConsoleMessage[] = []; + this.consoleMessages.forEach((messages, operatorId) => + messages.toArray().forEach(message => all.push({ operatorId, message })) + ); + return all.sort( + (a, b) => + a.message.timestamp.seconds - b.message.timestamp.seconds || + a.message.timestamp.nanos - b.message.timestamp.nanos + ); + } + hasConsoleMessages(operatorId: string): boolean { return this.consoleMessages.has(operatorId); } diff --git a/frontend/src/app/workspace/service/workflow-graph/model/joint-graph-wrapper.ts b/frontend/src/app/workspace/service/workflow-graph/model/joint-graph-wrapper.ts index acb31c4917f..44f0b5fe5eb 100644 --- a/frontend/src/app/workspace/service/workflow-graph/model/joint-graph-wrapper.ts +++ b/frontend/src/app/workspace/service/workflow-graph/model/joint-graph-wrapper.ts @@ -52,6 +52,8 @@ type JointLinkChangeEvent = [joint.dia.Link, { x: number; y: number }, { ui: boo type JointPositionChangeEvent = [joint.dia.Element, { x: number; y: number }]; +export const RESULT_PREVIEW_SETTING_KEY = "result-preview-on-hover"; + type PositionInfo = { currPos: Point; lastPos: Point | undefined; @@ -112,6 +114,10 @@ export class JointGraphWrapper { // main canvas and the mini-map. private heatmapViewStream = new BehaviorSubject(null); + // Whether hovering an operator previews its result (Layers > Result preview). On unless the + // user turned it off, which this browser remembers. + private resultPreviewEnabled = localStorage.getItem(RESULT_PREVIEW_SETTING_KEY) !== "false"; + private elementPositions: Map = new Map(); private listenPositionChange: boolean = true; @@ -247,6 +253,15 @@ export class JointGraphWrapper { return this.heatmapViewStream.asObservable(); } + public setResultPreviewEnabled(enabled: boolean): void { + this.resultPreviewEnabled = enabled; + localStorage.setItem(RESULT_PREVIEW_SETTING_KEY, String(enabled)); + } + + public isResultPreviewEnabled(): boolean { + return this.resultPreviewEnabled; + } + /** * This method is used to toggle the multiselect mode. * @param multiSelect 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..09270b06808 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); @@ -450,6 +485,47 @@ describe("OperatorPaginationResultService", () => { }); }); + describe("peekFirstRows", () => { + // A hover preview asks for the first rows while the result table may be on any page; the + // table's page and cache must come through untouched. + it("fetches the first rows without moving the table's page or reading its cache", () => { + const eventSubject = mockWorkflowWebsocketService.subscribeToEvent.mock.results[0] + .value as unknown as Subject; + (service as any).currentPageIndex = 4; + (service as any).resultCache.set(1, [{ id: "cached" }]); + const received: PaginatedResultEvent[] = []; + + service.peekFirstRows(5).subscribe(e => received.push(e)); + + expect(mockWorkflowWebsocketService.send).toHaveBeenCalledWith( + "ResultPaginationRequest", + expect.objectContaining({ + operatorID: "testOperator", + pageIndex: 1, + pageSize: 5, + columnOffset: 0, + // within the server's Int, which a larger number fails to parse as + columnLimit: 2147483647, + }) + ); + const requestID = (mockWorkflowWebsocketService.send.mock.calls[0][1] as any).requestID; + const page: PaginatedResultEvent = { + requestID, + operatorID: "testOperator", + pageIndex: 1, + table: [{ id: 1 }], + schema: [{ attributeName: "id", attributeType: "integer" }], + portIndex: 0, + totalTuples: 1, + }; + eventSubject.next(page); + + expect(received).toEqual([page]); + expect(service.getCurrentPageIndex()).toBe(4); + expect((service as any).resultCache.get(1)).toEqual([{ id: "cached" }]); + }); + }); + describe("handleResultUpdate", () => { it("records the total tuple count and evicts only the dirty pages", () => { (service as any).resultCache.set(1, [{ id: 1 }]); @@ -494,15 +570,77 @@ 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 every request, and the first port's service leaves it out", () => { + secondPort.selectPage(1, 10).subscribe(); + secondPort.peekFirstRows(5).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(sent[1].portIndex).toBe(1); + expect("portIndex" in sent[2]).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..e542b6856ea 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(); } @@ -239,6 +262,12 @@ export class OperatorResultService { } } +/** + * A column limit that asks for every column. The server reads the limit as a 32-bit Int, so it is + * Int.MaxValue there; Number.MAX_SAFE_INTEGER fails to parse and the request is never answered. + */ +export const ALL_COLUMNS = 2147483647; + export class OperatorPaginationResultService { private pendingRequests: Map> = new Map(); private resultCache: Map> = new Map(); @@ -249,10 +278,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 +329,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 +339,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 +354,7 @@ export class OperatorPaginationResultService { columnOffset, columnLimit, columnSearch, + ...this.portField(), }); const pendingRequestSubject = new Subject(); this.pendingRequests.set(requestID, pendingRequestSubject); @@ -327,6 +362,29 @@ export class OperatorPaginationResultService { } } + /** + * The first rows of the result, for a glance at it. Unlike selectPage this leaves the page the + * result table is on alone and never touches its cache, so a preview cannot move the table. + * Every column is asked for, so the schema the reply carries is the whole one, as a table + * page's would be. + */ + public peekFirstRows(rowCount: number): Observable { + const requestID = uuid(); + this.workflowWebsocketService.send("ResultPaginationRequest", { + requestID, + operatorID: this.operatorID, + pageIndex: 1, + pageSize: rowCount, + columnOffset: 0, + columnLimit: ALL_COLUMNS, + columnSearch: "", + ...this.portField(), + }); + const pendingRequestSubject = new Subject(); + this.pendingRequests.set(requestID, pendingRequestSubject); + return pendingRequestSubject; + } + public reset(): void { this.pendingRequests.clear(); this.resultCache.clear(); @@ -345,11 +403,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<{