diff --git a/amber/src/main/scala/org/apache/texera/web/model/websocket/event/PaginatedResultEvent.scala b/amber/src/main/scala/org/apache/texera/web/model/websocket/event/PaginatedResultEvent.scala
index 081d072180b..9f7983241e3 100644
--- a/amber/src/main/scala/org/apache/texera/web/model/websocket/event/PaginatedResultEvent.scala
+++ b/amber/src/main/scala/org/apache/texera/web/model/websocket/event/PaginatedResultEvent.scala
@@ -27,16 +27,31 @@ object PaginatedResultEvent {
def apply(
req: ResultPaginationRequest,
table: List[ObjectNode],
- schema: List[Attribute]
+ schema: List[Attribute],
+ totalTuples: Int = 0
): PaginatedResultEvent = {
- PaginatedResultEvent(req.requestID, req.operatorID, req.pageIndex, table, schema)
+ PaginatedResultEvent(
+ req.requestID,
+ req.operatorID,
+ req.pageIndex,
+ table,
+ schema,
+ req.portIndex,
+ totalTuples
+ )
}
}
+/**
+ * One page of an output port's result. `totalTuples` counts the whole port, so a port that the
+ * running execution does not report on, any but an operator's first, can still be paged through.
+ */
case class PaginatedResultEvent(
requestID: String,
operatorID: String,
pageIndex: Int,
table: List[ObjectNode],
- schema: List[Attribute]
+ schema: List[Attribute],
+ portIndex: Int,
+ totalTuples: Int
) extends TexeraWebSocketEvent
diff --git a/amber/src/main/scala/org/apache/texera/web/model/websocket/request/ResultPaginationRequest.scala b/amber/src/main/scala/org/apache/texera/web/model/websocket/request/ResultPaginationRequest.scala
index 4a3e0a58a3e..2696d90bcb3 100644
--- a/amber/src/main/scala/org/apache/texera/web/model/websocket/request/ResultPaginationRequest.scala
+++ b/amber/src/main/scala/org/apache/texera/web/model/websocket/request/ResultPaginationRequest.scala
@@ -26,5 +26,7 @@ case class ResultPaginationRequest(
pageSize: Int,
columnOffset: Int = 0,
columnLimit: Int = Int.MaxValue,
- columnSearch: Option[String] = None
+ columnSearch: Option[String] = None,
+ // which of the operator's output ports to read; an operator with one output has only port 0
+ portIndex: Int = 0
) extends TexeraWebSocketRequest
diff --git a/amber/src/main/scala/org/apache/texera/web/service/ExecutionResultService.scala b/amber/src/main/scala/org/apache/texera/web/service/ExecutionResultService.scala
index 2fa63f6b863..50d5c03de6b 100644
--- a/amber/src/main/scala/org/apache/texera/web/service/ExecutionResultService.scala
+++ b/amber/src/main/scala/org/apache/texera/web/service/ExecutionResultService.scala
@@ -478,7 +478,7 @@ class ExecutionResultService(
val storageUriOption = WorkflowExecutionsResource.getResultUriByLogicalPortId(
latestExecutionId,
OperatorIdentity(request.operatorID),
- PortIdentity()
+ PortIdentity(request.portIndex)
)
// Refuse to read a per-user-warehouse result while the feature is off (#6930).
@@ -511,7 +511,12 @@ class ExecutionResultService(
val attributes = paginationIterable.headOption
.map(_.getSchema.getAttributes)
.getOrElse(List.empty)
- PaginatedResultEvent.apply(request, mappedResults, attributes)
+ PaginatedResultEvent.apply(
+ request,
+ mappedResults,
+ attributes,
+ virtualDocument.getCount.toInt
+ )
case None =>
// Handle the case when storageUri is empty
@@ -524,6 +529,11 @@ class ExecutionResultService(
val newInfo: Map[OperatorIdentity, OperatorResultMetadata] = {
WorkflowExecutionsResource
.getResultUrisByExecutionId(executionId)
+ // One entry per operator, so it follows the first output port, as the result updates do.
+ // Keyed by operator, a second port's count would otherwise replace the first's.
+ .filter(uri =>
+ VFSURIFactory.decodeURI(uri).globalPortId.exists(_.portId == PortIdentity())
+ )
.map(uri => {
val count = DocumentFactory.openDocument(uri)._1.getCount.toInt
diff --git a/amber/src/test/scala/org/apache/texera/web/model/websocket/event/PaginatedResultEventSpec.scala b/amber/src/test/scala/org/apache/texera/web/model/websocket/event/PaginatedResultEventSpec.scala
index aa48cc81c8c..d07c881fa74 100644
--- a/amber/src/test/scala/org/apache/texera/web/model/websocket/event/PaginatedResultEventSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/web/model/websocket/event/PaginatedResultEventSpec.scala
@@ -60,7 +60,8 @@ class PaginatedResultEventSpec extends AnyFlatSpec with Matchers {
pageSize = 25,
columnOffset = 4,
columnLimit = 6,
- columnSearch = Some("city")
+ columnSearch = Some("city"),
+ portIndex = 1
)
private val table = List(
@@ -80,6 +81,12 @@ class PaginatedResultEventSpec extends AnyFlatSpec with Matchers {
event.pageIndex shouldBe 3
}
+ it should "echo the port the request read, and the port's row count" in {
+ val event = PaginatedResultEvent(request, table, schema, totalTuples = 40)
+ event.portIndex shouldBe 1
+ event.totalTuples shouldBe 40
+ }
+
it should "pass the table and schema through untouched" in {
val event = PaginatedResultEvent(request, table, schema)
event.table should contain theSameElementsInOrderAs table
@@ -88,10 +95,21 @@ class PaginatedResultEventSpec extends AnyFlatSpec with Matchers {
"the projected event" should "serialize to exactly the fields the Angular client declares" in {
val json = objectMapper.readTree(
- objectMapper.writeValueAsString(PaginatedResultEvent(request, table, schema))
+ objectMapper.writeValueAsString(PaginatedResultEvent(request, table, schema, 40))
)
json.fieldNames().asScala.toSet shouldBe
- Set("type", "requestID", "operatorID", "pageIndex", "table", "schema")
+ Set(
+ "type",
+ "requestID",
+ "operatorID",
+ "pageIndex",
+ "table",
+ "schema",
+ "portIndex",
+ "totalTuples"
+ )
+ json.get("portIndex").asInt() shouldBe 1
+ json.get("totalTuples").asInt() shouldBe 40
json.get("type").asText() shouldBe "PaginatedResultEvent"
json.get("requestID").asText() shouldBe "req-alpha"
json.get("operatorID").asText() shouldBe "op-beta"
diff --git a/amber/src/test/scala/org/apache/texera/web/model/websocket/event/TexeraWebSocketEventSpec.scala b/amber/src/test/scala/org/apache/texera/web/model/websocket/event/TexeraWebSocketEventSpec.scala
index 19dd3af6b7c..2aa250b3a7c 100644
--- a/amber/src/test/scala/org/apache/texera/web/model/websocket/event/TexeraWebSocketEventSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/web/model/websocket/event/TexeraWebSocketEventSpec.scala
@@ -147,7 +147,9 @@ class TexeraWebSocketEventSpec extends AnyFlatSpec with Matchers {
"op-2",
3,
List(resultRow),
- List(new Attribute("city", AttributeType.STRING))
+ List(new Attribute("city", AttributeType.STRING)),
+ 1,
+ 40
),
"PythonExpressionEvaluateResponse" -> PythonExpressionEvaluateResponse(
"len(tuple_)",
diff --git a/amber/src/test/scala/org/apache/texera/web/resource/WorkflowWebsocketResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/WorkflowWebsocketResourceSpec.scala
index 4761c5b3022..81057d74efa 100644
--- a/amber/src/test/scala/org/apache/texera/web/resource/WorkflowWebsocketResourceSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/web/resource/WorkflowWebsocketResourceSpec.scala
@@ -284,7 +284,9 @@ class WorkflowWebsocketResourceSpec
request.operatorID,
request.pageIndex,
List(objectMapper.createObjectNode().put("c", 5)),
- List(new Attribute("c", AttributeType.INTEGER))
+ List(new Attribute("c", AttributeType.INTEGER)),
+ request.portIndex,
+ 1
)
}
}
diff --git a/amber/src/test/scala/org/apache/texera/web/service/ExecutionResultServiceSpec.scala b/amber/src/test/scala/org/apache/texera/web/service/ExecutionResultServiceSpec.scala
index 3d9ed9fcf8a..0fec412ec5a 100644
--- a/amber/src/test/scala/org/apache/texera/web/service/ExecutionResultServiceSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/web/service/ExecutionResultServiceSpec.scala
@@ -925,6 +925,32 @@ class ExecutionResultServiceSpec
event.schema.map(_.getName) shouldBe List("c1", "c2")
}
+ it should "read the output port the request names, and count that port's rows" in {
+ val executionId = newExecution()
+ val schema = new Schema(List(new Attribute("id", AttributeType.INTEGER)))
+ val rows = (n: Int, offset: Int) =>
+ (0 until n).map(i =>
+ Tuple.builder(schema).add("id", AttributeType.INTEGER, offset + i).build()
+ )
+ // Two output ports of one operator, as a Split keeps them, with different row counts.
+ storeResult(executionId, "erss-two-ports", schema, rows(2, 0))
+ storeResult(executionId, "erss-two-ports", schema, rows(5, 100), port = PortIdentity(1))
+
+ val first = paginate(paginationRequest.copy(operatorID = "erss-two-ports", pageSize = 10))
+ val second = paginate(
+ paginationRequest.copy(operatorID = "erss-two-ports", pageSize = 3, portIndex = 1)
+ )
+
+ // Without a port, the first one is read, as before.
+ first.table.map(_.get("id").asInt()) shouldBe List(0, 1)
+ first.portIndex shouldBe 0
+ first.totalTuples shouldBe 2
+ // The second port's rows, and its own count, which pages the port past this first page.
+ second.table.map(_.get("id").asInt()) shouldBe List(100, 101, 102)
+ second.portIndex shouldBe 1
+ second.totalTuples shouldBe 5
+ }
+
it should "return an empty page when the operator has no stored result" in {
newExecution()
@@ -1237,6 +1263,34 @@ class ExecutionResultServiceSpec
}
}
+ it should "report an operator with two output ports by its first port's count" in {
+ val executionId = newExecution()
+ val schema = new Schema(List(new Attribute("id", AttributeType.INTEGER)))
+ val rows = (n: Int) =>
+ (0 until n).map(i => Tuple.builder(schema).add("id", AttributeType.INTEGER, i).build())
+ storeResult(executionId, "erss-final-two-ports", schema, rows(3))
+ storeResult(executionId, "erss-final-two-ports", schema, rows(8), port = PortIdentity(1))
+ val plan = planOf(
+ physicalOp(
+ executionId,
+ "erss-final-two-ports",
+ PortIdentity() -> OutputMode.SET_SNAPSHOT,
+ PortIdentity(1) -> OutputMode.SET_SNAPSHOT
+ )
+ )
+
+ withAttached(plan, executionId) { fixture =>
+ setPollingCancellable(fixture.service, new RecordingCancellable)
+
+ fixture.client.fire(ExecutionStateUpdate(WorkflowAggregatedState.COMPLETED))
+
+ // The update for an operator reads its first port, so its count has to come from there
+ // too; the second port's 8 rows must not stand in for the first port's 3.
+ val resultInfo = fixture.workflowStateStore.resultStore.getState.resultInfo
+ resultInfo(OperatorIdentity("erss-final-two-ports")).tupleCount shouldBe 3
+ }
+ }
+
it should "leave polling running while the execution has not terminated" in {
val executionId = newExecution()
withAttached(planOf(), executionId) { fixture =>
@@ -1450,10 +1504,13 @@ class ExecutionResultServiceSpec
ExecutionIdentity(execution.getEid.longValue())
}
- private def globalPortIdOf(operatorId: String): GlobalPortIdentity =
+ private def globalPortIdOf(
+ operatorId: String,
+ port: PortIdentity = PortIdentity()
+ ): GlobalPortIdentity =
GlobalPortIdentity(
PhysicalOpIdentity(OperatorIdentity(operatorId), "main"),
- PortIdentity(),
+ port,
input = false
)
@@ -1461,13 +1518,14 @@ class ExecutionResultServiceSpec
private def resultUriOf(
executionId: ExecutionIdentity,
operatorId: String,
- warehouse: Option[String] = None
+ warehouse: Option[String] = None,
+ port: PortIdentity = PortIdentity()
): URI =
VFSURIFactory.resultURI(
VFSURIFactory.createPortBaseURI(
WorkflowIdentity(testWid.longValue()),
executionId,
- globalPortIdOf(operatorId),
+ globalPortIdOf(operatorId, port),
warehouse
)
)
@@ -1492,9 +1550,10 @@ class ExecutionResultServiceSpec
executionId: ExecutionIdentity,
operatorId: String,
schema: Schema,
- tuples: Seq[Tuple]
+ tuples: Seq[Tuple],
+ port: PortIdentity = PortIdentity()
): URI = {
- val uri = resultUriOf(executionId, operatorId)
+ val uri = resultUriOf(executionId, operatorId, port = port)
val document = DocumentFactory.createDocument(uri, schema)
val writer = document.writer("erss").asInstanceOf[BufferedItemWriter[Tuple]]
writer.open()
diff --git a/frontend/src/app/workspace/component/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
+
+
+