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 4af3a85bcd test(amber): trim subsumed loop-body e2e cases to speed up
amber-integration (#6884)
4af3a85bcd is described below
commit 4af3a85bcd4ecd3c1f8b5588ed4a1e25361f7ef7
Author: Xinyuan Lin <[email protected]>
AuthorDate: Sat Jul 25 19:54:37 2026 -0700
test(amber): trim subsumed loop-body e2e cases to speed up
amber-integration (#6884)
### What changes were proposed in this PR?
`LoopIntegrationSpec` costs **~4m16s of the ~6-minute
`amber-integration` test step (~70%)** on every PR. The cost tracks
worker spawns almost perfectly (~2s per Python worker boot + Flight
handshake; 132 spawns across the suite), because every loop iteration
respawns each worker in the re-executed regions:
| Test | Time | Worker spawns (py + jvm) |
|---|---:|---:|
| single loop (Python-only) | 17s | 6 + 1 |
| nested 3×3 (Python-only) | 63s | 24 + 1 |
| single + Limit | 19s | 6 + 4 |
| nested + Limit | 64s | 24 + 10 |
| nested + chain (Limit→Sleep) | 72s | 24 + 19 |
| single + chain | 21s | 6 + 7 |
Two cases are **strict subsets** of the nested-chain case, which already
covers nested routing, counter magnitude across JVM hops, the JVM→JVM
state handoff, and both operator kinds in one workflow:
- `nested + Limit` (~64s) — nested × Limit ⊂ nested × (Limit → Sleep)
- `single + chain` (~21s) — chain ⊂ nested chain
This PR removes those two, saving **~85s (~33% of the loop suite)** with
no unique coverage lost. Retained matrix: Python-only single (base),
Python-only nested (pins the no-JVM path), single + Limit (verbatim
repro of the #6660 bug), nested + chain (superset JVM case). The
subsumption reasoning is recorded in the retained test's comment so the
cases don't get re-added.
The deeper fix (reusing workers across loop iterations instead of
respawn-per-jump) is an engine design change and out of scope here.
### Any related issues, documentation, discussions?
Follow-up to #6661 (which introduced the JVM loop-body cases). Related
engine context: #6660.
### How was this PR tested?
Test-only change (removes two e2e cases; no source touched).
`WorkflowExecutionService/Test/compile` + `scalafmtCheckAll` +
`scalafixAll --check` pass locally (Java 17); the remaining suite runs
in the `amber-integration` CI job on this PR.
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Fable 5)
---
.../amber/engine/e2e/LoopIntegrationSpec.scala | 78 +++-------------------
1 file changed, 11 insertions(+), 67 deletions(-)
diff --git
a/amber/src/test/integration/org/apache/texera/amber/engine/e2e/LoopIntegrationSpec.scala
b/amber/src/test/integration/org/apache/texera/amber/engine/e2e/LoopIntegrationSpec.scala
index 2c81f78045..3dec510b90 100644
---
a/amber/src/test/integration/org/apache/texera/amber/engine/e2e/LoopIntegrationSpec.scala
+++
b/amber/src/test/integration/org/apache/texera/amber/engine/e2e/LoopIntegrationSpec.scala
@@ -282,54 +282,22 @@ class LoopIntegrationSpec
)
}
- it should "run a nested loop whose inner body contains a JVM (Scala)
operator" in {
- // TextInput -> OuterStart -> InnerStart -> Limit -> InnerEnd -> OuterEnd.
- //
- // The nested variant of the JVM-hop test: the OUTER loop's state crosses
- // the Scala operator stamped (loop_counter = 1, loop_start_id = outer), so
- // this pins that the JVM hop preserves the counter MAGNITUDE too, not just
- // the id. A zeroed counter would make the inner LoopEnd mis-consume the
- // outer state (KeyError on the missing 'table' key / clobbered jump id)
- // instead of passing it through with -1.
- val src = textInput("1\n2\n3")
- val outerStart = loopStart("i = 0", "table")
- val innerStart = loopStart("j = 0", "table.iloc[j]")
- val mid = limit(10)
- val innerEnd = loopEnd("j += 1", "j < len(table)")
- val outerEnd = loopEnd("i += 1", "i < len(table)")
- val materialized = runAndGetMaterializedRowCounts(
- List(src, outerStart, innerStart, mid, innerEnd, outerEnd),
- List(
- link(src, outerStart),
- link(outerStart, innerStart),
- link(innerStart, mid),
- link(mid, innerEnd),
- link(innerEnd, outerEnd)
- )
- )
- val outerRows = materialized.getOrElse(outerEnd.operatorIdentifier, -1L)
- val innerRows = materialized.getOrElse(innerEnd.operatorIdentifier, -1L)
- assert(
- outerRows == 9,
- s"outer LoopEnd must accumulate all 9 inner-iteration rows with a Scala
" +
- s"operator in the inner body: expected 9, got $outerRows (all:
$materialized)"
- )
- assert(
- innerRows == 3,
- s"inner LoopEnd must reset per outer iteration (3 rows, not 9): " +
- s"expected 3, got $innerRows (all: $materialized)"
- )
- }
-
it should "run a nested loop whose inner body chains multiple JVM operators"
in {
// TextInput -> OuterStart -> InnerStart -> Limit -> Sleep(0) -> InnerEnd
-> OuterEnd.
//
// The strongest combination: nested (the OUTER loop's state crosses the
// JVM hops stamped with loop_counter = 1, so the counter MAGNITUDE must
- // survive both hops) x multi-hop (the JVM-to-JVM state handoff between
- // the two Scala workers) x both operator kinds. A zeroed counter or a
- // blanked id at either hop would mis-route the outer state at the inner
- // LoopEnd.
+ // survive both hops -- a zeroed counter would make the inner LoopEnd
+ // mis-consume the outer state instead of passing it through with -1) x
+ // multi-hop (the JVM-to-JVM state handoff: the first Scala worker WRITES
+ // the envelope columns to its output state table and the second Scala
+ // worker READS them back) x both operator kinds.
+ //
+ // This case strictly subsumes a nested loop with a single JVM operator
+ // and a single loop with a JVM chain; those were removed as separate
+ // tests because each loop iteration respawns every worker in the
+ // re-executed regions (~2s per Python worker), making every extra e2e
+ // workflow expensive (~1 minute per nested case in CI).
val src = textInput("1\n2\n3")
val outerStart = loopStart("i = 0", "table")
val innerStart = loopStart("j = 0", "table.iloc[j]")
@@ -363,28 +331,4 @@ class LoopIntegrationSpec
)
}
- it should "run a loop whose body chains multiple JVM operators" in {
- // TextInput -> LoopStart -> Limit -> Sleep(0) -> LoopEnd.
- //
- // Two consecutive JVM hops pin the JVM-to-JVM state handoff: the first
- // Scala worker WRITES the envelope columns to its output state table and
- // the second Scala worker READS them back. The single-JVM-op tests never
- // exercise that pairing (there, each hop faces a Python worker on at
- // least one side). Sleep(0) is a pure identity pass-through.
- val src = textInput("1\n2\n3")
- val start = loopStart("i = 0", "table.iloc[i]")
- val first = limit(10)
- val second = sleep(0)
- val end = loopEnd("i += 1", "i < len(table)")
- val materialized = runAndGetMaterializedRowCounts(
- List(src, start, first, second, end),
- List(link(src, start), link(start, first), link(first, second),
link(second, end))
- )
- val endRows = materialized.getOrElse(end.operatorIdentifier, -1L)
- assert(
- endRows == 3,
- s"LoopEnd must accumulate all 3 iterations with two chained JVM " +
- s"operators in the loop body: expected 3, got $endRows (all:
$materialized)"
- )
- }
}
