This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-6884-57d4dba20ae395ccd76795308cda8b190c3c1df9 in repository https://gitbox.apache.org/repos/asf/texera.git
commit 03da010a40caeb22d7b0ca09be65e732e60f9d32 Author: Xinyuan Lin <[email protected]> AuthorDate: Sat Jul 25 19:17:23 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)" - ) - } }
