This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/texera.git
The following commit(s) were added to refs/heads/main by this push:
new a3f2bf0a1d test(amber): cover result pagination and the result update
loop (#7556)
a3f2bf0a1d is described below
commit a3f2bf0a1db064c7dd710a1fb6b10bc2832f94dd
Author: Xinyuan Lin <[email protected]>
AuthorDate: Tue Aug 11 22:06:11 2026 -0700
test(amber): cover result pagination and the result update loop (#7556)
### What changes were proposed in this PR?
`ExecutionResultService` sat at **22.9% of 122 lines**. Its 18 existing
tests cover the JSON conversion helpers and the `WebOutputMode`
round-trips and stop at the class's own behaviour, so nothing exercised
the paths a user actually hits: paging through a stored result,
searching and slicing its columns, and the polling loop that pushes
updates to the frontend while an execution runs.
Adds 22 tests to the existing spec, taking the file to **99.2% of
lines** (121/122).
The seam is `attachToExecution`'s `client` parameter: `AmberClient` is
non-final with an overridable `registerCallback`, so a test subclass
captures the registrations and fires them directly over a bare
`ActorSystem` — the pattern `ExecutionConsoleServiceSpec` already uses.
A fresh `ExecutionStateStore` that never sees RUNNING is what keeps
`AmberRuntime` out of it.
Covered: page origin and range end, case-insensitive column search,
column offset and limit, the warehouse read guard, all three output
modes and the internal-port filter, the dirty-page computation, snapshot
versus delta reads, table statistics, the terminal-state transition that
cancels polling and runs one final update, and the fatal-error path.
### Verification
32 mutations applied one at a time, each reverted with the production
diff confirmed empty before the next. All 32 red. Three were then re-run
independently after the fact — the page origin off-by-one page, the
dirty-page count flooring instead of ceiling, and the delta reading from
the new tuple count instead of the old — all three red again.
Two details worth stating rather than glossing:
- **The page end bound rests on one assertion.** Only the exact-id-list
test pins it; the "clamp the last page" test cannot, because with 7 rows
both `[6,9)` and `[6,10)` yield the same single row. The redundancy one
would assume is not there, so that test is load-bearing on its own.
- **The warehouse-guard test uses `a[WarehouseUnavailableException]
should be thrownBy`**, which is a loose form — any collaborator throwing
that type satisfies it, and `DocumentFactory.openDocument` is a
plausible second thrower since it also resolves warehouses. Dropping the
guard was checked directly and fails exactly that one test, so the line
is genuinely load-bearing for it.
### Deliberately not included
One line remains uncovered, and it should be **deleted rather than
tested**: the `case _ => throw new RuntimeException("update mode
combination not supported: ...")` in `convertWebResultUpdate`.
`webOutputMode` is built immediately above from a total match over
`OutputMode`, so it is provably one of `PaginationMode` /
`SetSnapshotMode` / `SetDeltaMode` and all three are matched by the
preceding cases. No test can kill a mutation there.
No production file is touched.
### Any related issues, documentation, discussions?
Closes #7555
### How was this PR tested?
```
STORAGE_ICEBERG_CATALOG_TYPE=postgres sbt
"WorkflowExecutionService/testOnly
org.apache.texera.web.service.ExecutionResultServiceSpec"
```
```
[info] Total number of tests run: 40
[info] Tests: succeeded 40, failed 0, canceled 0, ignored 0, pending 0
```
22 new on top of the existing 18. Coverage measured with sbt-jacoco
filtered to this spec — note that a plain `testOnly` reports 0% for this
module, since the destfile javaOption only comes from the `jacoco` task.
`Test/scalafmtCheck` and `Test/scalafix --check` both pass.
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 5)
---
.../web/service/ExecutionResultServiceSpec.scala | 857 ++++++++++++++++++++-
1 file changed, 854 insertions(+), 3 deletions(-)
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 e355c8aef0..73c69886ec 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
@@ -21,14 +21,36 @@ package org.apache.texera.web.service
import com.fasterxml.jackson.databind.JsonNode
import com.fasterxml.jackson.databind.node.ObjectNode
+import io.reactivex.rxjava3.disposables.Disposable
+import org.apache.pekko.actor.{ActorSystem, Cancellable}
+import org.apache.texera.amber.core.executor.OpExecInitInfo
+import org.apache.texera.amber.core.storage.model.BufferedItemWriter
+import org.apache.texera.amber.core.storage.result.{OperatorResultMetadata,
WorkflowResultStore}
+import org.apache.texera.amber.core.storage.{DocumentFactory, VFSURIFactory}
import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema,
Tuple}
import org.apache.texera.amber.core.virtualidentity.{
ExecutionIdentity,
OperatorIdentity,
- PhysicalOpIdentity
+ PhysicalOpIdentity,
+ WorkflowIdentity
}
-import org.apache.texera.amber.core.workflow.{GlobalPortIdentity, PortIdentity}
-import
org.apache.texera.amber.engine.architecture.coordinator.OperatorPortResultUriAvailable
+import org.apache.texera.amber.core.workflow.OutputPort.OutputMode
+import org.apache.texera.amber.core.workflow.{
+ GlobalPortIdentity,
+ OutputPort,
+ PhysicalOp,
+ PhysicalPlan,
+ PortIdentity,
+ WorkflowContext
+}
+import org.apache.texera.amber.engine.architecture.coordinator.{
+ CoordinatorConfig,
+ ExecutionStateUpdate,
+ FatalError,
+ OperatorPortResultUriAvailable
+}
+import
org.apache.texera.amber.engine.architecture.rpc.controlreturns.WorkflowAggregatedState
+import org.apache.texera.amber.engine.common.client.AmberClient
import org.apache.texera.amber.util.JSONUtils.objectMapper
import org.apache.texera.amber.util.serde.GlobalPortIdentitySerde.SerdeOps
import org.apache.texera.dao.MockTexeraDB
@@ -36,11 +58,14 @@ import org.apache.texera.dao.jooq.generated.Tables.{
OPERATOR_PORT_EXECUTIONS,
USER,
WORKFLOW,
+ WORKFLOW_COMPUTING_UNIT,
WORKFLOW_EXECUTIONS,
WORKFLOW_VERSION
}
+import org.apache.texera.dao.jooq.generated.enums.WorkflowComputingUnitTypeEnum
import org.apache.texera.dao.jooq.generated.tables.daos.{
UserDao,
+ WorkflowComputingUnitDao,
WorkflowDao,
WorkflowExecutionsDao,
WorkflowVersionDao
@@ -48,9 +73,16 @@ import org.apache.texera.dao.jooq.generated.tables.daos.{
import org.apache.texera.dao.jooq.generated.tables.pojos.{
User,
Workflow,
+ WorkflowComputingUnit,
WorkflowExecutions,
WorkflowVersion
}
+import org.apache.texera.web.model.websocket.event.{
+ PaginatedResultEvent,
+ TexeraWebSocketEvent,
+ WebResultUpdateEvent
+}
+import org.apache.texera.web.model.websocket.request.ResultPaginationRequest
import org.apache.texera.web.service.ExecutionResultService.{
PaginationMode,
SetDeltaMode,
@@ -58,13 +90,20 @@ import
org.apache.texera.web.service.ExecutionResultService.{
WebDataUpdate,
WebPaginationUpdate
}
+import org.apache.texera.web.storage.{ExecutionStateStore, WorkflowStateStore}
import org.scalatest.flatspec.AnyFlatSpec
import org.scalatest.matchers.should.Matchers
import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach}
import java.net.URI
import java.sql.Timestamp
+import scala.collection.immutable.ListMap
+import scala.collection.mutable
import scala.jdk.CollectionConverters._
+import scala.reflect.ClassTag
+import java.util.UUID
+import scala.concurrent.Await
+import scala.concurrent.duration.DurationInt
class ExecutionResultServiceSpec
extends AnyFlatSpec
@@ -79,12 +118,24 @@ class ExecutionResultServiceSpec
private val testUid: Integer = 9001
private var executionsDao: WorkflowExecutionsDao = _
private var testVid: Integer = _
+ private var testCuid: Integer = _
+
+ // AmberClient needs an ActorSystem to host its ClientActor. A bare one is
enough:
+ // the client is constructed over an empty PhysicalPlan, so its
InitializeRequest
+ // completes without an engine, and the subclass below never sends anything
to it.
+ private var system: ActorSystem = _
override protected def beforeAll(): Unit = {
initializeDBAndReplaceDSLContext()
+ // Suffixed so two ActorSystems can never contend for the same name if
this spec is ever run
+ // alongside another in one JVM.
+ system = ActorSystem(s"ExecutionResultServiceSpec-${UUID.randomUUID()}")
}
override protected def afterAll(): Unit = {
+ // terminate() is asynchronous; without the await the JVM can move on to
shutdownDB() while the
+ // dispatcher threads are still running, which is how this kind of fixture
starts hanging.
+ Await.ready(system.terminate(), 30.seconds)
shutdownDB()
}
@@ -95,6 +146,18 @@ class ExecutionResultServiceSpec
user.setEmail(s"[email protected]")
new UserDao(getDSLContext.configuration()).insert(user)
+ // getLatestExecutionId matches on cuid, and a NULL cuid matches no value,
so
+ // handleResultPagination can only find an execution that names a real
unit.
+ val computingUnit = new WorkflowComputingUnit
+ computingUnit.setUid(testUid)
+ computingUnit.setName("execution-result-test-unit")
+ computingUnit.setCreationTime(new Timestamp(System.currentTimeMillis()))
+ computingUnit.setType(WorkflowComputingUnitTypeEnum.local)
+ computingUnit.setUri("local://execution-result-test")
+ computingUnit.setResource("{}")
+ new
WorkflowComputingUnitDao(getDSLContext.configuration()).insert(computingUnit)
+ testCuid = computingUnit.getCuid
+
val workflow = new Workflow
workflow.setWid(testWid)
workflow.setName(s"execution-result-test-$testWid")
@@ -116,7 +179,13 @@ class ExecutionResultServiceSpec
executionsDao = new WorkflowExecutionsDao(getDSLContext.configuration())
}
+ /** Every ResultEvents subscription made during a test, disposed when that
test ends. */
+ private val openObservers = mutable.ArrayBuffer.empty[Disposable]
+
override protected def afterEach(): Unit = {
+ openObservers.foreach(_.dispose())
+ openObservers.clear()
+
val ctx = getDSLContext
// Scope every delete to the test's own ids so this spec stays safe
// if it ever shares a DB with another spec.
@@ -137,6 +206,7 @@ class ExecutionResultServiceSpec
.execute()
ctx.deleteFrom(WORKFLOW_VERSION).where(WORKFLOW_VERSION.WID.eq(testWid)).execute()
ctx.deleteFrom(WORKFLOW).where(WORKFLOW.WID.eq(testWid)).execute()
+
ctx.deleteFrom(WORKFLOW_COMPUTING_UNIT).where(WORKFLOW_COMPUTING_UNIT.UID.eq(testUid)).execute()
ctx.deleteFrom(USER).where(USER.UID.eq(testUid)).execute()
}
@@ -696,4 +766,785 @@ class ExecutionResultServiceSpec
table.size() shouldBe 1
table.get(0).get("k").asText() shouldBe "v"
}
+
+ // ==========================================================================
+ // The service instance: handleResultPagination, and the callbacks / result
+ // diff handler that attachToExecution wires up.
+ //
+ // Two seams make this reachable without an engine:
+ // * `client` is a parameter of attachToExecution and `registerCallback` is
+ // overridable, so `RecordingAmberClient` captures the callbacks the
+ // service registers and each test fires one directly.
+ // * the result-polling `Cancellable` is created by AmberRuntime's
scheduler
+ // once the execution reaches RUNNING. AmberRuntime holds its ActorSystem
+ // in a private var that only a started pekko cluster fills in, so the
+ // service's own private field is set directly (`setPollingCancellable`)
+ // to reproduce the post-RUNNING state that the terminal-state,
fatal-error
+ // and re-attach paths all assume. Nothing global is mutated, and no
+ // RUNNING state is ever pushed through the metadata store.
+ //
+ // The result documents are real Iceberg tables. Their storage key is derived
+ // from the URI path, and creating one overwrites whatever is already there,
so
+ // every logical operator id below carries an `erss-` prefix that is unique
in
+ // the repository -- sbt runs amber suites in parallel inside one JVM.
+ // ==========================================================================
+
+ private val paginationRequest =
+ ResultPaginationRequest(requestID = "req-1", operatorID = "", pageIndex =
1, pageSize = 3)
+
+ "handleResultPagination" should "return the rows of the requested page" in {
+ val executionId = newExecution()
+ val schema = new Schema(
+ List(new Attribute("id", AttributeType.INTEGER), new Attribute("label",
AttributeType.STRING))
+ )
+ storeResult(
+ executionId,
+ "erss-page-rows",
+ schema,
+ (0 until 7).map(i =>
+ Tuple
+ .builder(schema)
+ .add("id", AttributeType.INTEGER, i)
+ .add("label", AttributeType.STRING, s"row-$i")
+ .build()
+ )
+ )
+
+ val event = paginate(
+ paginationRequest.copy(operatorID = "erss-page-rows", pageIndex = 2,
pageSize = 3)
+ )
+
+ // Page 2 of size 3 is the half-open range [3, 6): both the `pageSize *
+ // (pageIndex - 1)` origin and the `from + pageSize` bound are pinned by
the
+ // exact id list, which shifts under an off-by-one on either end.
+ event.table.map(_.get("id").asInt()) shouldBe List(3, 4, 5)
+ event.table.map(_.get("label").asText()) shouldBe List("row-3", "row-4",
"row-5")
+ // The request fields are echoed back so the frontend can match the reply
to
+ // the outstanding page request.
+ event.requestID shouldBe "req-1"
+ event.operatorID shouldBe "erss-page-rows"
+ event.pageIndex shouldBe 2
+ event.schema.map(_.getName) shouldBe List("id", "label")
+ }
+
+ it should "clamp the last page to the rows that exist" in {
+ val executionId = newExecution()
+ val schema = new Schema(List(new Attribute("id", AttributeType.INTEGER)))
+ storeResult(
+ executionId,
+ "erss-page-rows",
+ schema,
+ (0 until 7).map(i => Tuple.builder(schema).add("id",
AttributeType.INTEGER, i).build())
+ )
+
+ val event = paginate(
+ paginationRequest.copy(operatorID = "erss-page-rows", pageIndex = 3,
pageSize = 3)
+ )
+
+ event.table.map(_.get("id").asInt()) shouldBe List(6)
+ }
+
+ it should "return an empty page with no schema past the end of the result"
in {
+ val executionId = newExecution()
+ val schema = new Schema(List(new Attribute("id", AttributeType.INTEGER)))
+ storeResult(
+ executionId,
+ "erss-page-rows",
+ schema,
+ (0 until 7).map(i => Tuple.builder(schema).add("id",
AttributeType.INTEGER, i).build())
+ )
+
+ val event = paginate(
+ paginationRequest.copy(operatorID = "erss-page-rows", pageIndex = 4,
pageSize = 3)
+ )
+
+ event.table shouldBe empty
+ // The reported schema comes from the first row of the page, so a page
with no
+ // rows reports no schema at all -- there is no fallback to the stored
schema.
+ event.schema shouldBe empty
+ }
+
+ it should "match a column search case-insensitively" in {
+ val executionId = newExecution()
+ val schema = new Schema(
+ List(
+ new Attribute("userName", AttributeType.STRING),
+ new Attribute("UserAge", AttributeType.STRING),
+ new Attribute("city", AttributeType.STRING)
+ )
+ )
+ storeResult(
+ executionId,
+ "erss-col-search",
+ schema,
+ List(
+ Tuple
+ .builder(schema)
+ .add("userName", AttributeType.STRING, "ada")
+ .add("UserAge", AttributeType.STRING, "36")
+ .add("city", AttributeType.STRING, "london")
+ .build()
+ )
+ )
+
+ // "eRNa" matches "userName" only when BOTH sides are lower-cased: dropping
+ // the column's `.toLowerCase` leaves "userName".contains("erna") false,
and
+ // dropping the search term's leaves "username".contains("eRNa") false.
+ val event = paginate(
+ paginationRequest
+ .copy(operatorID = "erss-col-search", columnSearch = Some("eRNa"))
+ )
+
+ event.table.map(_.fieldNames().asScala.toList) shouldBe
List(List("userName"))
+ event.table.head.get("userName").asText() shouldBe "ada"
+ event.schema.map(_.getName) shouldBe List("userName")
+ }
+
+ it should "slice the projected columns by offset and limit" in {
+ val executionId = newExecution()
+ val columnNames = List("c0", "c1", "c2", "c3")
+ val schema = new Schema(columnNames.map(new Attribute(_,
AttributeType.STRING)))
+ storeResult(
+ executionId,
+ "erss-col-slice",
+ schema,
+ List(
+ columnNames
+ .foldLeft(Tuple.builder(schema))((b, c) => b.add(c,
AttributeType.STRING, s"v-$c"))
+ .build()
+ )
+ )
+
+ // slice's second argument is an end index, not a count: `slice(offset,
limit)`
+ // would return only "c1" here.
+ val event = paginate(
+ paginationRequest.copy(operatorID = "erss-col-slice", columnOffset = 1,
columnLimit = 2)
+ )
+
+ event.table.map(_.fieldNames().asScala.toList) shouldBe List(List("c1",
"c2"))
+ event.schema.map(_.getName) shouldBe List("c1", "c2")
+ }
+
+ it should "return an empty page when the operator has no stored result" in {
+ newExecution()
+
+ val event = paginate(
+ paginationRequest.copy(operatorID = "erss-never-stored", pageIndex = 2)
+ )
+
+ // The reply still has to identify the request it answers, otherwise the
+ // frontend cannot retire the pending page.
+ event.requestID shouldBe "req-1"
+ event.operatorID shouldBe "erss-never-stored"
+ event.pageIndex shouldBe 2
+ event.table shouldBe empty
+ event.schema shouldBe empty
+ }
+
+ it should "fail loudly when the workflow has never been executed" in {
+ // No execution row at all: distinct from the case above, where an
execution
+ // exists but stored no result for the operator.
+ val thrown = the[IllegalStateException] thrownBy paginate(
+ paginationRequest.copy(operatorID = "erss-page-rows")
+ )
+ thrown.getMessage shouldBe "No execution is recorded"
+ }
+
+ it should "refuse to read a result stored in a per-user warehouse while the
feature is off" in {
+ val executionId = newExecution()
+ // Only the URI row is needed: the guard has to reject before the document
is
+ // opened, and no table was ever created for this warehouse. The guard
reads
+ // StorageConfig.warehouseEnabled, which ships (and runs in CI) as false;
+ // WarehouseReadGuardSpec pins both settings of the flag directly.
+ insertResultUri(executionId, resultUriOf(executionId, "erss-warehouse",
Some("byo")))
+
+ a[WarehouseUnavailableException] should be thrownBy paginate(
+ paginationRequest.copy(operatorID = "erss-warehouse")
+ )
+ }
+
+ // -- attachToExecution: the result-store diff handler
-----------------------
+
+ "the result diff handler" should "send a pagination update for a
SET_SNAPSHOT output port" in {
+ val executionId = newExecution()
+ val schema = new Schema(List(new Attribute("id", AttributeType.INTEGER)))
+ storeResult(
+ executionId,
+ "erss-snapshot",
+ schema,
+ (0 until 7).map(i => Tuple.builder(schema).add("id",
AttributeType.INTEGER, i).build())
+ )
+ // The internal port is listed FIRST so that dropping the
`!portId.internal`
+ // filter would read SET_DELTA's mode instead of the external port's.
+ val plan = planOf(
+ physicalOp(
+ executionId,
+ "erss-snapshot",
+ PortIdentity(1, internal = true) -> OutputMode.SET_DELTA,
+ PortIdentity() -> OutputMode.SET_SNAPSHOT
+ )
+ )
+
+ withAttached(plan, executionId) { fixture =>
+ val events = new ResultEvents(fixture.workflowStateStore)
+ fixture.setResultCounts("erss-snapshot" -> 7)
+
+ // 7 rows at the default page size of 5 dirty two pages; the count sent
to
+ // the frontend is the new count, not the old one.
+ events.updates.map(_.updates) shouldBe List(
+ Map("erss-snapshot" -> WebPaginationUpdate(PaginationMode(), 7L,
List(1, 2)))
+ )
+ }
+ }
+
+ it should "record the size and per-column statistics of a table-mode result"
in {
+ val executionId = newExecution()
+ val schema = new Schema(
+ List(new Attribute("id", AttributeType.INTEGER), new Attribute("name",
AttributeType.STRING))
+ )
+ val uri = storeResult(
+ executionId,
+ "erss-snapshot-stats",
+ schema,
+ List(
+ Tuple
+ .builder(schema)
+ .add("id", AttributeType.INTEGER, 1)
+ .add("name", AttributeType.STRING, "ada")
+ .build()
+ )
+ )
+ val plan = planOf(
+ physicalOp(executionId, "erss-snapshot-stats", PortIdentity() ->
OutputMode.SET_SNAPSHOT)
+ )
+
+ withAttached(plan, executionId) { fixture =>
+ val events = new ResultEvents(fixture.workflowStateStore)
+ fixture.setResultCounts("erss-snapshot-stats" -> 1)
+
+ events.updates.head.tableStats.keySet shouldBe Set("erss-snapshot-stats")
+ events.updates.head.tableStats("erss-snapshot-stats").keySet shouldBe
Set("id", "name")
+
+ // The persisted size must be the document's own file size, not some
other
+ // measure of the result (a row count, or a hard-coded zero).
+ // The persisted size must be the document's own file size, not some
other
+ // measure of the result (a row count, or a hard-coded zero). Nothing is
closed
+ // afterwards because VirtualDocument exposes no close(); its only
teardown is
+ // clear(), which deletes the data rather than releasing a handle.
+ val expectedSize = DocumentFactory.openDocument(uri)._1.getTotalFileSize
+ expectedSize should not be 0L
+ storedResultSize(executionId, "erss-snapshot-stats") shouldBe
expectedSize
+ }
+ }
+
+ it should "send the whole untruncated snapshot, and no statistics, for a
SINGLE_SNAPSHOT port" in {
+ val executionId = newExecution()
+ val schema = new Schema(List(new Attribute("html", AttributeType.STRING)))
+ val html = "a" * 150
+ // Two rows, so that reading only the head of the document would be
visible.
+ storeResult(
+ executionId,
+ "erss-single",
+ schema,
+ List(html, "<p>second</p>")
+ .map(Tuple.builder(schema).add("html", AttributeType.STRING,
_).build())
+ )
+ val plan =
+ planOf(physicalOp(executionId, "erss-single", PortIdentity() ->
OutputMode.SINGLE_SNAPSHOT))
+
+ withAttached(plan, executionId) { fixture =>
+ val events = new ResultEvents(fixture.workflowStateStore)
+ fixture.setResultCounts("erss-single" -> 2)
+
+ val update = dataUpdate(events.updates.head.updates("erss-single"))
+ update.mode shouldBe SetSnapshotMode()
+ // SINGLE_SNAPSHOT carries rendered HTML, so the 100-character string
+ // truncation must be switched off for it.
+ update.table.map(_.get("html").asText()) shouldBe List(html,
"<p>second</p>")
+ // A SINGLE_SNAPSHOT port's document is not a table, so no statistics are
+ // gathered and no size is recorded for it -- while the data update
itself
+ // is still sent.
+ events.updates.head.tableStats shouldBe empty
+ storedResultSize(executionId, "erss-single") shouldBe 0L
+ }
+ }
+
+ it should "send only the rows added since the previous update for a
SET_DELTA port" in {
+ val executionId = newExecution()
+ val schema = new Schema(
+ List(new Attribute("id", AttributeType.INTEGER), new Attribute("note",
AttributeType.STRING))
+ )
+ val longNote = "b" * 150
+ storeResult(
+ executionId,
+ "erss-delta",
+ schema,
+ (0 until 5).map(i =>
+ Tuple
+ .builder(schema)
+ .add("id", AttributeType.INTEGER, i)
+ .add("note", AttributeType.STRING, longNote)
+ .build()
+ )
+ )
+ val plan = planOf(physicalOp(executionId, "erss-delta", PortIdentity() ->
OutputMode.SET_DELTA))
+
+ withAttached(plan, executionId) { fixture =>
+ val events = new ResultEvents(fixture.workflowStateStore)
+
+ fixture.setResultCounts("erss-delta" -> 2)
+ fixture.setResultCounts("erss-delta" -> 5)
+
+ val tables =
+ events.updates.map(e =>
dataUpdate(e.updates("erss-delta")).table.map(_.get("id").asInt()))
+ // The delta is read from the PREVIOUS count, so the second update
repeats
+ // nothing: reading from the new count would send an empty table, and
+ // reading from zero would resend rows 0 and 1.
+ tables shouldBe List(List(0, 1, 2, 3, 4), List(2, 3, 4))
+
+ val deltaUpdate = dataUpdate(events.updates.head.updates("erss-delta"))
+ deltaUpdate.mode shouldBe SetDeltaMode()
+ // Unlike SINGLE_SNAPSHOT, a delta is table data, so long strings are
+ // truncated for display.
+ deltaUpdate.table.head.get("note").asText() shouldBe ("b" * 100 + "...")
+ }
+ }
+
+ it should "send an empty pagination update when the result is not stored
yet" in {
+ val executionId = newExecution()
+ val plan =
+ planOf(physicalOp(executionId, "erss-unstored", PortIdentity() ->
OutputMode.SET_SNAPSHOT))
+
+ withAttached(plan, executionId) { fixture =>
+ val events = new ResultEvents(fixture.workflowStateStore)
+ fixture.setResultCounts("erss-unstored" -> 4)
+
+ // Storage only exists once the operator's region has been scheduled.
Until
+ // then the reported count is zero, not the count the engine reported.
+ events.updates.map(_.updates) shouldBe List(
+ Map("erss-unstored" -> WebPaginationUpdate(PaginationMode(), 0L,
List.empty))
+ )
+ }
+ }
+
+ it should "reject an output mode it does not recognize" in {
+ val executionId = newExecution()
+ val plan = planOf(
+ physicalOp(executionId, "erss-bad-mode", PortIdentity() ->
OutputMode.Unrecognized(99))
+ )
+
+ withAttached(plan, executionId) { fixture =>
+ val events = new ResultEvents(fixture.workflowStateStore)
+ fixture.setResultCounts("erss-bad-mode" -> 1)
+
+ events.updates shouldBe empty
+ // The unrecognized enum value renders as "UNRECOGNIZED", so the
workflow id
+ // is the only thing in the message that identifies what failed.
+ events.errors.map(_.getMessage) shouldBe List(
+ s"Unrecognized output mode: UNRECOGNIZED for workflow $testWid"
+ )
+ }
+ }
+
+ it should "update only the operators whose tuple count changed" in {
+ val executionId = newExecution()
+ val schema = new Schema(List(new Attribute("id", AttributeType.INTEGER)))
+ val rows = List(Tuple.builder(schema).add("id", AttributeType.INTEGER,
1).build())
+ storeResult(executionId, "erss-changed-a", schema, rows)
+ storeResult(executionId, "erss-changed-b", schema, rows)
+ val plan = planOf(
+ physicalOp(executionId, "erss-changed-a", PortIdentity() ->
OutputMode.SET_SNAPSHOT),
+ physicalOp(executionId, "erss-changed-b", PortIdentity() ->
OutputMode.SET_SNAPSHOT)
+ )
+
+ withAttached(plan, executionId) { fixture =>
+ val events = new ResultEvents(fixture.workflowStateStore)
+ fixture.setResultCounts("erss-changed-a" -> 1, "erss-changed-b" -> 1)
+ fixture.setResultCounts("erss-changed-a" -> 1, "erss-changed-b" -> 2)
+
+ // Recomputing an unchanged operator would re-read its whole document on
+ // every poll, so the second diff must mention only the operator that
moved.
+ events.updates.map(_.updates.keySet) shouldBe List(
+ Set("erss-changed-a", "erss-changed-b"),
+ Set("erss-changed-b")
+ )
+ }
+ }
+
+ // -- attachToExecution: the engine callbacks -------------------------------
+
+ "the OperatorPortResultUriAvailable callback" should
+ "persist the URI under the attached execution" in {
+ val executionId = newExecution()
+ withAttached(planOf(), executionId) { fixture =>
+ val globalPortId = globalPortIdOf("erss-callback")
+ val uri = resultUriOf(executionId, "erss-callback")
+
+ fixture.client.fire(OperatorPortResultUriAvailable(globalPortId, uri))
+
+ val rows = getDSLContext
+ .selectFrom(OPERATOR_PORT_EXECUTIONS)
+
.where(OPERATOR_PORT_EXECUTIONS.WORKFLOW_EXECUTION_ID.eq(executionId.id.toInt))
+ .fetch()
+ rows.size() shouldBe 1
+ rows.get(0).getGlobalPortId shouldBe globalPortId.serializeAsString
+ rows.get(0).getResultUri shouldBe uri.toString
+ }
+ }
+
+ "the ExecutionStateUpdate callback" should
+ "stop polling and run one final update when the execution terminates" in {
+ val executionId = newExecution()
+ val schema = new Schema(List(new Attribute("id", AttributeType.INTEGER)))
+ storeResult(
+ executionId,
+ "erss-final-snapshot",
+ schema,
+ (0 until 3).map(i => Tuple.builder(schema).add("id",
AttributeType.INTEGER, i).build())
+ )
+ storeResult(
+ executionId,
+ "erss-final-delta",
+ schema,
+ (0 until 2).map(i => Tuple.builder(schema).add("id",
AttributeType.INTEGER, i).build())
+ )
+ val plan = planOf(
+ physicalOp(executionId, "erss-final-snapshot", PortIdentity() ->
OutputMode.SET_SNAPSHOT),
+ physicalOp(executionId, "erss-final-delta", PortIdentity() ->
OutputMode.SET_DELTA)
+ )
+
+ withAttached(plan, executionId) { fixture =>
+ val polling = new RecordingCancellable
+ setPollingCancellable(fixture.service, polling)
+
+
fixture.client.fire(ExecutionStateUpdate(WorkflowAggregatedState.COMPLETED))
+
+ polling.cancelCount shouldBe 1
+ val resultInfo =
fixture.workflowStateStore.resultStore.getState.resultInfo
+ resultInfo.keySet.map(_.id) shouldBe Set("erss-final-snapshot",
"erss-final-delta")
+ // Counts come from the documents themselves, so the two operators must
not
+ // be reported with the same count.
+ resultInfo(OperatorIdentity("erss-final-snapshot")).tupleCount shouldBe 3
+ resultInfo(OperatorIdentity("erss-final-delta")).tupleCount shouldBe 2
+ // A SET_SNAPSHOT result is replaced wholesale, so its content can change
+ // without its count changing; the random change detector is what makes
the
+ // frontend re-read it. A delta is append-only and needs no such nudge.
+ resultInfo(OperatorIdentity("erss-final-snapshot")).changeDetector
should not be empty
+ resultInfo(OperatorIdentity("erss-final-delta")).changeDetector shouldBe
""
+ }
+ }
+
+ it should "leave polling running while the execution has not terminated" in {
+ val executionId = newExecution()
+ withAttached(planOf(), executionId) { fixture =>
+ val polling = new RecordingCancellable
+ setPollingCancellable(fixture.service, polling)
+
+ fixture.client.fire(ExecutionStateUpdate(WorkflowAggregatedState.PAUSED))
+
+ polling.cancelCount shouldBe 0
+ fixture.workflowStateStore.resultStore.getState.resultInfo shouldBe empty
+ }
+ }
+
+ "the FatalError callback" should "stop polling" in {
+ val executionId = newExecution()
+ withAttached(planOf(), executionId) { fixture =>
+ val polling = new RecordingCancellable
+ setPollingCancellable(fixture.service, polling)
+
+ fixture.client.fire(FatalError(new RuntimeException("boom")))
+
+ polling.cancelCount shouldBe 1
+ }
+ }
+
+ it should "tolerate a fatal error raised before polling ever started" in {
+ val executionId = newExecution()
+ withAttached(planOf(), executionId) { fixture =>
+ // A workflow can fail during initialization, i.e. before it reaches
RUNNING
+ // and creates a polling handle. Without the null guard this is an NPE
+ // inside the error path, which would mask the original failure.
+ noException should be thrownBy fixture.client.fire(FatalError(new
RuntimeException("boom")))
+ }
+ }
+
+ "attachToExecution" should "cancel the previous execution's polling" in {
+ val executionId = newExecution()
+ withAttached(planOf(), executionId) { fixture =>
+ val polling = new RecordingCancellable
+ setPollingCancellable(fixture.service, polling)
+
+ fixture.service.attachToExecution(
+ executionId,
+ new ExecutionStateStore,
+ planOf(),
+ fixture.client
+ )
+
+ // Two cancels, from two different places: attachToExecution's own guard,
+ // and then the freshly registered metadata subscription, which replays
the
+ // store's current (non-RUNNING) state immediately and cancels on it. The
+ // next case pins the guard by removing its contribution.
+ polling.cancelCount shouldBe 2
+ }
+ }
+
+ it should "not re-cancel polling that is already cancelled" in {
+ val executionId = newExecution()
+ withAttached(planOf(), executionId) { fixture =>
+ val polling = new RecordingCancellable(initiallyCancelled = true)
+ setPollingCancellable(fixture.service, polling)
+
+ fixture.service.attachToExecution(
+ executionId,
+ new ExecutionStateStore,
+ planOf(),
+ fixture.client
+ )
+
+ // Only the metadata subscription's cancel is left: attachToExecution's
guard
+ // skips a handle that reports itself already cancelled.
+ polling.cancelCount shouldBe 1
+ }
+ }
+
+ // ---------------------------------------------------------------- fixtures
+
+ /** An AmberClient that records the callbacks the service registers, so
tests can fire them. */
+ private final class RecordingAmberClient
+ extends AmberClient(
+ system,
+ new WorkflowContext(),
+ PhysicalPlan(Set.empty, Set.empty),
+ CoordinatorConfig(None, None, None, None),
+ _ => ()
+ ) {
+ private val callbacks = mutable.Map.empty[Class[_], Any => Unit]
+
+ override def registerCallback[T](callback: T => Unit)(implicit ct:
ClassTag[T]): Disposable = {
+ callbacks(ct.runtimeClass) = callback.asInstanceOf[Any => Unit]
+ Disposable.empty()
+ }
+
+ /** Delivers `event` to the callback registered for its type; fails if
there is none. */
+ def fire[T <: AnyRef](event: T): Unit =
+ callbacks
+ .getOrElse(
+ event.getClass,
+ fail(s"no callback registered for ${event.getClass.getSimpleName}")
+ )
+ .apply(event)
+ }
+
+ /** A Cancellable that counts cancellations, standing in for AmberRuntime's
scheduled poll. */
+ private final class RecordingCancellable(initiallyCancelled: Boolean = false)
+ extends Cancellable {
+
+ /** Counts calls, not successful cancellations -- the tests assert how
often production asks. */
+ var cancelCount = 0
+ private var cancelled = initiallyCancelled
+
+ // Follows Cancellable's contract: true only if THIS call did the
cancelling. Returning true
+ // unconditionally would quietly diverge from the real scheduler if
production ever branches on
+ // the result.
+ override def cancel(): Boolean = {
+ cancelCount += 1
+ val didCancel = !cancelled
+ cancelled = true
+ didCancel
+ }
+
+ override def isCancelled: Boolean = cancelled
+ }
+
+ private final class Fixture(
+ val client: RecordingAmberClient,
+ val service: ExecutionResultService,
+ val workflowStateStore: WorkflowStateStore
+ ) {
+
+ /** Publishes one engine-reported tuple count per operator, as the poller
does. */
+ def setResultCounts(counts: (String, Int)*): Unit =
+ workflowStateStore.resultStore.updateState(_ =>
+ WorkflowResultStore(counts.map {
+ case (opId, count) => OperatorIdentity(opId) ->
OperatorResultMetadata(count)
+ }.toMap)
+ )
+ }
+
+ /** Collects what the result store's diff handler emits, and any error it
raises. */
+ private final class ResultEvents(workflowStateStore: WorkflowStateStore) {
+ private val collected = mutable.ArrayBuffer.empty[TexeraWebSocketEvent]
+ val errors: mutable.ArrayBuffer[Throwable] =
mutable.ArrayBuffer.empty[Throwable]
+
+ // Held and disposed in afterEach: an observer left subscribed keeps
collecting into this
+ // buffer after its test has finished, and retains the buffer with it.
+ private val subscription: Disposable =
+ workflowStateStore.resultStore.getWebsocketEventObservable.subscribe(
+ (evts: Iterable[TexeraWebSocketEvent]) => collected ++= evts,
+ (err: Throwable) => errors += err
+ )
+
+ openObservers += subscription
+
+ def updates: List[WebResultUpdateEvent] =
+ collected.collect { case e: WebResultUpdateEvent => e }.toList
+ }
+
+ private def withAttached(physicalPlan: PhysicalPlan, executionId:
ExecutionIdentity)(
+ body: Fixture => Unit
+ ): Unit = {
+ val client = new RecordingAmberClient
+ val workflowStateStore = new WorkflowStateStore
+ val service = new ExecutionResultService(
+ WorkflowIdentity(testWid.longValue()),
+ testCuid,
+ workflowStateStore
+ )
+ try {
+ service.attachToExecution(executionId, new ExecutionStateStore,
physicalPlan, client)
+ body(new Fixture(client, service, workflowStateStore))
+ } finally {
+ service.unsubscribeAll()
+ client.shutdown()
+ }
+ }
+
+ /**
+ * Runs `request` against a service bound to the seeded workflow and
computing
+ * unit, so getLatestExecutionId resolves against the rows this spec
inserts.
+ */
+ private def paginate(request: ResultPaginationRequest): PaginatedResultEvent
= {
+ val service = new ExecutionResultService(
+ WorkflowIdentity(testWid.longValue()),
+ testCuid,
+ new WorkflowStateStore
+ )
+ service.handleResultPagination(request) match {
+ case event: PaginatedResultEvent => event
+ case other => fail(s"expected a
PaginatedResultEvent, got $other")
+ }
+ }
+
+ private def dataUpdate(update: ExecutionResultService.WebResultUpdate):
WebDataUpdate =
+ update match {
+ case dataUpdate: WebDataUpdate => dataUpdate
+ case other => fail(s"expected a WebDataUpdate, got
$other")
+ }
+
+ private def newExecution(): ExecutionIdentity = {
+ val execution = new WorkflowExecutions
+ execution.setVid(testVid)
+ execution.setUid(testUid)
+ execution.setCuid(testCuid)
+ execution.setStatus(0.toByte)
+ execution.setStartingTime(new Timestamp(System.currentTimeMillis()))
+ execution.setBookmarked(false)
+ execution.setName("execution-result-instance-test")
+ execution.setEnvironmentVersion("test-env")
+ executionsDao.insert(execution)
+ ExecutionIdentity(execution.getEid.longValue())
+ }
+
+ private def globalPortIdOf(operatorId: String): GlobalPortIdentity =
+ GlobalPortIdentity(
+ PhysicalOpIdentity(OperatorIdentity(operatorId), "main"),
+ PortIdentity(),
+ input = false
+ )
+
+ /** The external-output result URI shape that getResultUriByLogicalPortId
decodes and matches. */
+ private def resultUriOf(
+ executionId: ExecutionIdentity,
+ operatorId: String,
+ warehouse: Option[String] = None
+ ): URI =
+ VFSURIFactory.resultURI(
+ VFSURIFactory.createPortBaseURI(
+ WorkflowIdentity(testWid.longValue()),
+ executionId,
+ globalPortIdOf(operatorId),
+ warehouse
+ )
+ )
+
+ private def insertResultUri(executionId: ExecutionIdentity, uri: URI): Unit =
+ getDSLContext
+ .insertInto(OPERATOR_PORT_EXECUTIONS)
+ .columns(
+ OPERATOR_PORT_EXECUTIONS.WORKFLOW_EXECUTION_ID,
+ OPERATOR_PORT_EXECUTIONS.GLOBAL_PORT_ID,
+ OPERATOR_PORT_EXECUTIONS.RESULT_URI
+ )
+ .values(
+ Integer.valueOf(executionId.id.toInt),
+ VFSURIFactory.decodeURI(uri).globalPortId.get.serializeAsString,
+ uri.toString
+ )
+ .execute()
+
+ /** Creates a real Iceberg result table holding `tuples`, and records its
URI for `executionId`. */
+ private def storeResult(
+ executionId: ExecutionIdentity,
+ operatorId: String,
+ schema: Schema,
+ tuples: Seq[Tuple]
+ ): URI = {
+ val uri = resultUriOf(executionId, operatorId)
+ val document = DocumentFactory.createDocument(uri, schema)
+ val writer =
document.writer("erss").asInstanceOf[BufferedItemWriter[Tuple]]
+ writer.open()
+ tuples.foreach(writer.putOne)
+ writer.close()
+ insertResultUri(executionId, uri)
+ uri
+ }
+
+ private def storedResultSize(executionId: ExecutionIdentity, operatorId:
String): Long =
+ getDSLContext
+ .select(OPERATOR_PORT_EXECUTIONS.RESULT_SIZE)
+ .from(OPERATOR_PORT_EXECUTIONS)
+
.where(OPERATOR_PORT_EXECUTIONS.WORKFLOW_EXECUTION_ID.eq(executionId.id.toInt))
+ .and(
+
OPERATOR_PORT_EXECUTIONS.GLOBAL_PORT_ID.eq(globalPortIdOf(operatorId).serializeAsString)
+ )
+ .fetchOne(OPERATOR_PORT_EXECUTIONS.RESULT_SIZE)
+
+ /**
+ * A single-layer physical op whose output ports are declared in the given
order --
+ * a ListMap, because convertWebResultUpdate reads the FIRST external
port's mode.
+ */
+ private def physicalOp(
+ executionId: ExecutionIdentity,
+ operatorId: String,
+ ports: (PortIdentity, OutputMode)*
+ ): PhysicalOp =
+ PhysicalOp(
+ id = PhysicalOpIdentity(OperatorIdentity(operatorId), "main"),
+ workflowId = WorkflowIdentity(testWid.longValue()),
+ executionId = executionId,
+ opExecInitInfo = OpExecInitInfo.Empty,
+ outputPorts = ListMap.from(ports.map {
+ case (portId, mode) =>
+ portId -> (OutputPort(id = portId, mode = mode), List.empty,
Right(Schema()))
+ })
+ )
+
+ private def planOf(ops: PhysicalOp*): PhysicalPlan = PhysicalPlan(ops.toSet,
Set.empty)
+
+ /**
+ * Sets the private polling handle. AmberRuntime creates it once the
execution
+ * reaches RUNNING, from an ActorSystem it keeps in a private var that only
a
+ * started pekko cluster fills in; the terminal-state, fatal-error and
re-attach
+ * paths are all defined by what they do to an already-created handle.
+ */
+ private def setPollingCancellable(
+ service: ExecutionResultService,
+ cancellable: Cancellable
+ ): Unit = {
+ val field =
classOf[ExecutionResultService].getDeclaredField("resultUpdateCancellable")
+ field.setAccessible(true)
+ field.set(service, cancellable)
+ }
}
