Skip to content
Original file line number Diff line number Diff line change
Expand Up @@ -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
Original file line number Diff line number Diff line change
Expand Up @@ -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
Original file line number Diff line number Diff line change
Expand Up @@ -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).
Expand Down Expand Up @@ -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
Expand All @@ -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

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand All @@ -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
Expand All @@ -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"
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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_)",
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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
)
}
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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()

Expand Down Expand Up @@ -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 =>
Expand Down Expand Up @@ -1462,24 +1516,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
)
)
Expand All @@ -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()
Expand Down
8 changes: 8 additions & 0 deletions frontend/src/app/workspace/component/menu/menu.component.html
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
6 changes: 6 additions & 0 deletions frontend/src/app/workspace/component/menu/menu.component.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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()
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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)),
}),
};
Expand Down Expand Up @@ -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 });
Expand Down Expand Up @@ -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" };
Expand Down Expand Up @@ -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 } },
],
Expand Down Expand Up @@ -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 } },
],
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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> = {};

Expand Down Expand Up @@ -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 => {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -86,7 +86,7 @@ <h4>Empty result set</h4>
</th>
</tr>
<tr
*ngIf="tableStats && prevTableStats"
*ngIf="hasTableStats"
#statsRow
class="custom-stats-row">
<th *ngFor="let column of currentColumns">
Expand Down
Loading
Loading