feat(frontend): preview an operator's result when hovering it on the canvas

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

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

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
 

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"

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_)",

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
       )
     }
   }

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,24 +1504,28 @@ 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
     )
 
   /** The external-output result URI shape that getResultUriByLogicalPortId decodes and matches. */
   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()

frontend/src/app/workspace/component/menu/menu.component.html

@@ -227,6 +227,14 @@
                   >Status</label
                 >
               </li>
+              <li nz-menu-item>
+                <label
+                  nz-checkbox
+                  [(ngModel)]="showResultPreview"
+                  (ngModelChange)="toggleResultPreview()"
+                  >Result preview</label
+                >
+              </li>
               <li nz-menu-item>
                 <label
                   nz-checkbox

frontend/src/app/workspace/component/menu/menu.component.ts

@@ -160,6 +160,7 @@ export class MenuComponent implements OnInit, OnDestroy {
   public showNumWorkers: boolean = false;
   public showStatus: boolean = false;
   public showHeatmap: boolean = false;
+  public showResultPreview: boolean = true;
   public heatmapView: HeatmapView = HeatmapView.Runtime;
   public HeatmapView = HeatmapView; // make Angular HTML access enum definition
   protected readonly USER_WORKFLOW = USER_WORKFLOW;
@@ -243,6 +244,7 @@ export class MenuComponent implements OnInit, OnDestroy {
 
   public ngOnInit(): void {
     this.restorePersistedHeatmapOverlay();
+    this.showResultPreview = this.workflowActionService.getJointGraphWrapper().isResultPreviewEnabled();
     // The export flags are reset when a menu is destroyed, which is right when the workspace is
     // left and wrong when a workflow's two views hand over and the results are kept. Recompute
     // from what is in hand, so a menu arriving on retained results does not offer a dead button.
@@ -353,6 +355,10 @@ export class MenuComponent implements OnInit, OnDestroy {
     document.body.removeChild(tempSpan);
   }
 
+  toggleResultPreview() {
+    this.workflowActionService.getJointGraphWrapper().setResultPreviewEnabled(this.showResultPreview);
+  }
+
   toggleNumWorkers() {
     this.workflowActionService
       .getJointGraphWrapper()

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 } },
       ],

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<string, unknown> } =
-    inject(NZ_MODAL_DATA);
+  private readonly modalData: {
+    operatorId: string;
+    rowIndex: number;
+    rowData?: Record<string, unknown>;
+    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<string, unknown> = {};
 
@@ -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 => {