You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@texera.apache.org by "aglinxinyuan (via GitHub)" <gi...@apache.org> on 2026/07/05 03:03:24 UTC
[PR] test(amber): add X-shaped multi-region workflow e2e integration test [texera]
aglinxinyuan opened a new pull request, #6132:
URL: https://github.com/apache/texera/pull/6132
### What changes were proposed in this PR?
Add `MultiRegionWorkflowIntegrationSpec`, an end-to-end integration test for a workflow that executes across **multiple regions**.
Cross-region data delivery goes through input-port materialization reader threads (which replaced the cache-source operators in #3425). Until now that path had no end-to-end coverage — only scheduling unit tests that count regions on a physical plan. This spec adds that coverage with a big "X"-shaped workflow:
```
csvLeft ─┬─▶ join.build (port 0) ─┐
│ join ─▶ pythonSort ─▶ (terminal)
csvRight ─┼─▶ join.probe (port 1) ─┘
│
csvLeft ─┼─▶ union ───────────────────────────────────▶ (terminal)
csvRight ─┘ (two links fan into union's single port)
```
The hash join's probe-depends-on-build ordering forces the build input to be materialized, cutting the plan into ≥2 regions whose boundary is crossed by the reader-thread path. The test asserts both structure and correctness:
- the plan spans **≥2 regions** (via the same `CostBasedScheduleGenerator` the controller uses at runtime);
- the self-join on the unique `Order ID` key yields **100 rows**, the Python UDF returns them **sorted**, and the union concatenates both sources into **200 rows**.
A Python UDF (sort) makes this a real end-to-end run with worker subprocesses, so it is class-level `@IntegrationTest` tagged and routed to the `amber-integration` CI job (which provisions Python deps); the lighter `amber` job excludes it.
### Any related issues, documentation, discussions?
Lands ahead of #6126 (removal of the now-dead `CacheSourceOpExec` / `OpExecSource` plumbing) so the multi-region path has end-to-end coverage **before** that refactor. Requested by @Yicong-Huang during review of #6126.
### How was this PR tested?
- `sbt "WorkflowExecutionService/Test/compile"` — green
- `sbt "WorkflowExecutionService/Test/scalafmtCheck" "WorkflowExecutionService/scalafixAll --check"` — clean
- The spec itself runs in the `amber-integration` CI job (it spawns Python workers, so it cannot run on Windows locally). Verified passing there: the run spawned the `WF*-HashJoinOpDesc-build`/`-probe`, `WF*-UnionOpDesc`, and `WF*-PythonUDFOpDescV2` workers and all tests passed.
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8 [1M context])
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(amber): add X-shaped multi-region workflow e2e integration test [texera]
Posted by "Yicong-Huang (via GitHub)" <gi...@apache.org>.
Yicong-Huang commented on code in PR #6132:
URL: https://github.com/apache/texera/pull/6132#discussion_r3524376440
##########
amber/src/test/integration/org/apache/texera/amber/engine/e2e/MultiRegionWorkflowIntegrationSpec.scala:
##########
@@ -0,0 +1,291 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.amber.engine.e2e
+
+import com.twitter.util.{Await, Duration, Promise, Return, Throw}
+import com.typesafe.scalalogging.Logger
+import org.apache.pekko.actor.{ActorSystem, Props}
+import org.apache.pekko.testkit.{ImplicitSender, TestKit}
+import org.apache.pekko.util.Timeout
+import org.apache.texera.amber.clustering.SingleNodeListener
+import org.apache.texera.amber.core.workflow.PortIdentity
+import org.apache.texera.amber.engine.architecture.controller.{
+ ControllerConfig,
+ ExecutionStateUpdate,
+ FatalError
+}
+import org.apache.texera.amber.engine.architecture.rpc.controlcommands.EmptyRequest
+import org.apache.texera.amber.engine.architecture.rpc.controlreturns.WorkflowAggregatedState.COMPLETED
+import org.apache.texera.amber.engine.architecture.scheduling.CostBasedScheduleGenerator
+import org.apache.texera.amber.engine.common.AmberRuntime
+import org.apache.texera.amber.engine.common.client.AmberClient
+import org.apache.texera.amber.engine.common.virtualidentity.util.CONTROLLER
+import org.apache.texera.amber.engine.e2e.TestUtils.{
+ buildWorkflow,
+ cleanupWorkflowExecutionData,
+ initiateTexeraDBForTestCases,
+ runWorkflowAndReadTerminalResults,
+ setUpWorkflowExecutionData
+}
+import org.apache.texera.amber.operator.TestOperators
+import org.apache.texera.amber.operator.source.scan.text.TextInputSourceOpDesc
+import org.apache.texera.amber.operator.union.UnionOpDesc
+import org.apache.texera.amber.tags.IntegrationTest
+import org.apache.texera.workflow.LogicalLink
+import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach, Outcome, Retries}
+import org.scalatest.flatspec.AnyFlatSpecLike
+
+import scala.concurrent.duration._
+
+/**
+ * End-to-end coverage for a workflow that executes across MULTIPLE regions.
+ *
+ * Cross-region data used to be delivered by dedicated cache-source operators;
+ * #3425 replaced them with input-port materialization reader threads. Before
+ * removing the now-dead cache-source plumbing, this spec adds the missing
+ * end-to-end coverage for that multi-region path: until now it was exercised
+ * only by scheduling unit tests (region counting on a physical plan), so a
+ * regression in the materialization read/write path could pass CI unnoticed.
+ *
+ * This spec drives a big "X"-shaped workflow to guard that path:
+ *
+ * {{{
+ * csvLeft ─┬─▶ join.build (port 0) ─┐
+ * │ join ─▶ pythonSort ─▶ (terminal)
+ * csvRight ─┼─▶ join.probe (port 1) ─┘
+ * │
+ * csvLeft ─┼─▶ union ───────────────────────────────────▶ (terminal)
+ * csvRight ─┘ (two links fan into union's single port)
+ * }}}
+ *
+ * The hash join's probe-depends-on-build ordering forces the build input to be
+ * materialized, cutting the plan into >=2 regions whose boundary is crossed by
+ * the reader-thread path. A Python UDF (sort) makes it a real end-to-end run
+ * with worker subprocesses, so it is class-level `@IntegrationTest` tagged and
+ * routed to the `amber-integration` CI job (which provisions Python deps); the
+ * lighter `amber` job excludes this tag.
+ */
+@IntegrationTest
+class MultiRegionWorkflowIntegrationSpec
+ extends TestKit(ActorSystem("MultiRegionWorkflowIntegrationSpec", AmberRuntime.pekkoConfig))
+ with ImplicitSender
+ with AnyFlatSpecLike
+ with BeforeAndAfterAll
+ with BeforeAndAfterEach
+ with Retries {
+
+ /**
+ * Retry each test once if it fails. Mirrors the other e2e integration specs:
+ * in CI there is a small chance the run does not observe "COMPLETED", so a
+ * single retry stabilizes the suite until that root cause is fixed.
+ */
+ override def withFixture(test: NoArgTest): Outcome =
+ withRetry { super.withFixture(test) }
+
+ implicit val timeout: Timeout = Timeout(5.seconds)
+
+ private val logger = Logger("MultiRegionWorkflowIntegrationSpecLogger")
+ private val specId = 5
+ private val ctx = TestUtils.workflowContext(specId)
+
+ // country_sales_small.csv has 100 rows with a unique "Order ID" per row.
+ private val sourceRowCount = 100
+
+ // Buffer every input tuple, then emit them ordered by "Order ID" once the
+ // input port is exhausted. This makes the UDF a genuine sort and forces a real
+ // Python worker to process data that crossed a region boundary.
+ private val sortByOrderIdCode =
+ """
+ |from pytexera import *
+ |
+ |class ProcessTupleOperator(UDFOperatorV2):
+ | def open(self) -> None:
+ | self.buffer = []
+ |
+ | @overrides
+ | def process_tuple(self, tuple_: Tuple, port: int) -> Iterator[Optional[TupleLike]]:
+ | self.buffer.append(tuple_)
+ | yield from []
+ |
+ | @overrides
+ | def on_finish(self, port: int) -> Iterator[Optional[TupleLike]]:
+ | for row in sorted(self.buffer, key=lambda t: t["Order ID"]):
+ | yield row
+ |""".stripMargin
+
+ override protected def beforeEach(): Unit = {
+ setUpWorkflowExecutionData(specId)
+ }
+
+ override protected def afterEach(): Unit = {
+ cleanupWorkflowExecutionData(specId)
+ }
+
+ override def beforeAll(): Unit = {
+ system.actorOf(Props[SingleNodeListener](), "cluster-info")
+ // These test cases access postgres in CI, but occasionally the jdbc driver cannot be found during CI run.
+ // Explicitly load the JDBC driver to avoid flaky CI failures.
+ Class.forName("org.postgresql.Driver")
+ initiateTexeraDBForTestCases()
+ warmupOnce()
+ }
+
+ override def afterAll(): Unit = {
+ TestKit.shutdownActorSystem(system)
+ }
+
+ /**
+ * Runs a TextInput -> Python UDF workflow once before the timed test so the
+ * Python worker cold-start is paid here, not inside the timed run. Capped and
+ * wrapped so warmup can never fail or hang the suite.
+ */
+ private def warmupOnce(): Unit = {
+ val warmupCap = Duration.fromSeconds(60)
+ setUpWorkflowExecutionData(specId)
+ var client: AmberClient = null
+ try {
+ val src = new TextInputSourceOpDesc()
+ src.textInput = "warmup"
+ val udf = TestOperators.pythonOpDesc()
+ val warmupCtx = TestUtils.workflowContext(specId)
+ val workflow = buildWorkflow(
+ List(src, udf),
+ List(
+ LogicalLink(
+ src.operatorIdentifier,
+ PortIdentity(),
+ udf.operatorIdentifier,
+ PortIdentity()
+ )
+ ),
+ warmupCtx
+ )
+ // Fail the completion Promise on any error (client-reported or FatalError)
+ // so a broken warmup returns immediately instead of blocking the whole
+ // warmupCap; the surrounding try/catch then logs and lets tests run cold.
+ val completion = Promise[Unit]()
+ client = new AmberClient(
+ system,
+ workflow.context,
+ workflow.physicalPlan,
+ ControllerConfig.default,
+ e => completion.updateIfEmpty(Throw(e))
+ )
+ client.registerCallback[FatalError](evt => completion.updateIfEmpty(Throw(evt.e)))
+ client.registerCallback[ExecutionStateUpdate](evt => {
+ if (evt.state == COMPLETED) completion.updateIfEmpty(Return(()))
+ })
+ Await.result(
+ client.controllerInterface.startWorkflow(EmptyRequest(), ()),
+ warmupCap
+ )
+ Await.result(completion, warmupCap)
+ } catch {
+ case e: Throwable =>
+ logger.warn(
+ s"warmup workflow did not finish within ${warmupCap}; tests will run cold and rely on Retries: ${e.getMessage}"
+ )
+ } finally {
+ if (client != null) {
+ try client.shutdown()
+ catch { case _: Throwable => () }
+ }
+ cleanupWorkflowExecutionData(specId)
+ }
+ }
+
+ "Engine" should "execute an X-shaped multi-region workflow (join + union + python) end-to-end" in {
+ val csvLeft = TestOperators.smallCsvScanOpDesc()
+ val csvRight = TestOperators.smallCsvScanOpDesc()
+ // Self-join on the unique "Order ID" key: each row matches exactly one row
+ // from the other source, so the join emits exactly `sourceRowCount` rows.
+ val join = TestOperators.joinOpDesc("Order ID", "Order ID")
+ val pythonSort = TestOperators.pythonOpDesc()
+ pythonSort.code = sortByOrderIdCode
+ val union = new UnionOpDesc()
+
+ val workflow = buildWorkflow(
+ List(csvLeft, csvRight, join, pythonSort, union),
+ List(
+ // Left arm of the X: the two sources feed the join (build/probe) ...
+ LogicalLink(
+ csvLeft.operatorIdentifier,
+ PortIdentity(),
+ join.operatorIdentifier,
+ PortIdentity()
+ ),
+ LogicalLink(
+ csvRight.operatorIdentifier,
+ PortIdentity(),
+ join.operatorIdentifier,
+ PortIdentity(1)
+ ),
+ LogicalLink(
+ join.operatorIdentifier,
+ PortIdentity(),
+ pythonSort.operatorIdentifier,
+ PortIdentity()
+ ),
+ // ... and also fan into the union's single input port (the crossing).
+ LogicalLink(
+ csvLeft.operatorIdentifier,
+ PortIdentity(),
+ union.operatorIdentifier,
+ PortIdentity()
+ ),
+ LogicalLink(
+ csvRight.operatorIdentifier,
+ PortIdentity(),
+ union.operatorIdentifier,
+ PortIdentity()
+ )
+ ),
+ ctx
+ )
+
+ // The join's build/probe dependency cuts the plan into multiple regions
+ // whose boundary is crossed by the input-port materialization reader path.
+ // Assert that with the same generator the controller uses at runtime.
+ val (schedule, _) = new CostBasedScheduleGenerator(
+ workflow.context,
+ workflow.physicalPlan,
+ CONTROLLER
+ ).generate()
+ assert(
+ schedule.getRegions.size >= 2,
+ s"expected the X-shaped workflow to span multiple regions, got ${schedule.getRegions.size}"
+ )
Review Comment:
Let's make it more end to end. e.g., from submitting a logical workflow. This way we can test the real source code, instead of calling different components inside. I hope this is an integration test, not unit test (so nothing to mock, or patch).
here we are using physical plan + manually wire up scheduler, which may drift from the real src code. I see you are doing this so that we can assert there are two regions created internally, for end to end test, we can skip the assertion (treating the execution like a black box), or use other methods to verify post execution.
##########
amber/src/test/integration/org/apache/texera/amber/engine/e2e/MultiRegionWorkflowIntegrationSpec.scala:
##########
@@ -0,0 +1,286 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.amber.engine.e2e
+
+import com.twitter.util.{Await, Duration, Promise, Return}
+import com.typesafe.scalalogging.Logger
+import org.apache.pekko.actor.{ActorSystem, Props}
+import org.apache.pekko.testkit.{ImplicitSender, TestKit}
+import org.apache.pekko.util.Timeout
+import org.apache.texera.amber.clustering.SingleNodeListener
+import org.apache.texera.amber.core.workflow.PortIdentity
+import org.apache.texera.amber.engine.architecture.controller.{
+ ControllerConfig,
+ ExecutionStateUpdate
+}
+import org.apache.texera.amber.engine.architecture.rpc.controlcommands.EmptyRequest
+import org.apache.texera.amber.engine.architecture.rpc.controlreturns.WorkflowAggregatedState.COMPLETED
+import org.apache.texera.amber.engine.architecture.scheduling.CostBasedScheduleGenerator
+import org.apache.texera.amber.engine.common.AmberRuntime
+import org.apache.texera.amber.engine.common.client.AmberClient
+import org.apache.texera.amber.engine.common.virtualidentity.util.CONTROLLER
+import org.apache.texera.amber.engine.e2e.TestUtils.{
+ buildWorkflow,
+ cleanupWorkflowExecutionData,
+ initiateTexeraDBForTestCases,
+ runWorkflowAndReadTerminalResults,
+ setUpWorkflowExecutionData
+}
+import org.apache.texera.amber.operator.TestOperators
+import org.apache.texera.amber.operator.source.scan.text.TextInputSourceOpDesc
+import org.apache.texera.amber.operator.union.UnionOpDesc
+import org.apache.texera.amber.tags.IntegrationTest
+import org.apache.texera.workflow.LogicalLink
+import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach, Outcome, Retries}
+import org.scalatest.flatspec.AnyFlatSpecLike
+
+import scala.concurrent.duration._
+
+/**
+ * End-to-end coverage for a workflow that executes across MULTIPLE regions.
+ *
+ * Cross-region data used to be delivered by dedicated cache-source operators;
+ * #3425 replaced them with input-port materialization reader threads. Before
+ * removing the now-dead cache-source plumbing, this spec adds the missing
+ * end-to-end coverage for that multi-region path: until now it was exercised
+ * only by scheduling unit tests (region counting on a physical plan), so a
+ * regression in the materialization read/write path could pass CI unnoticed.
+ *
+ * This spec drives a big "X"-shaped workflow to guard that path:
+ *
+ * {{{
+ * csvLeft ─┬─▶ join.build (port 0) ─┐
+ * │ join ─▶ pythonSort ─▶ (terminal)
+ * csvRight ─┼─▶ join.probe (port 1) ─┘
+ * │
+ * csvLeft ─┼─▶ union ───────────────────────────────────▶ (terminal)
+ * csvRight ─┘ (two links fan into union's single port)
+ * }}}
+ *
+ * The hash join's probe-depends-on-build ordering forces the build input to be
+ * materialized, cutting the plan into >=2 regions whose boundary is crossed by
+ * the reader-thread path. A Python UDF (sort) makes it a real end-to-end run
+ * with worker subprocesses, so it is class-level `@IntegrationTest` tagged and
+ * routed to the `amber-integration` CI job (which provisions Python deps); the
+ * lighter `amber` job excludes this tag.
+ */
+@IntegrationTest
+class MultiRegionWorkflowIntegrationSpec
+ extends TestKit(ActorSystem("MultiRegionWorkflowIntegrationSpec", AmberRuntime.pekkoConfig))
+ with ImplicitSender
+ with AnyFlatSpecLike
+ with BeforeAndAfterAll
+ with BeforeAndAfterEach
+ with Retries {
+
+ /**
+ * Retry each test once if it fails. Mirrors the other e2e integration specs:
+ * in CI there is a small chance the run does not observe "COMPLETED", so a
+ * single retry stabilizes the suite until that root cause is fixed.
+ */
+ override def withFixture(test: NoArgTest): Outcome =
+ withRetry { super.withFixture(test) }
+
+ implicit val timeout: Timeout = Timeout(5.seconds)
+
+ private val logger = Logger("MultiRegionWorkflowIntegrationSpecLogger")
+ private val specId = 5
+ private val ctx = TestUtils.workflowContext(specId)
+
+ // country_sales_small.csv has 100 rows with a unique "Order ID" per row.
+ private val sourceRowCount = 100
+
+ // Buffer every input tuple, then emit them ordered by "Order ID" once the
+ // input port is exhausted. This makes the UDF a genuine sort and forces a real
+ // Python worker to process data that crossed a region boundary.
+ private val sortByOrderIdCode =
Review Comment:
use the existing sort op? But good idea to have another UDF as well. May be add a Python source?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(amber): add X-shaped multi-region workflow e2e integration test [texera]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #6132:
URL: https://github.com/apache/texera/pull/6132#issuecomment-4884672233
## [Codecov](https://app.codecov.io/gh/apache/texera/pull/6132?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 59.12%. Comparing base ([`3a12e62`](https://app.codecov.io/gh/apache/texera/commit/3a12e6214c5f45477042374ae3a4a569835d1fda?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`a4b45d7`](https://app.codecov.io/gh/apache/texera/commit/a4b45d7207b3ce68be459dba6bfc6991f2b9c710?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:white_check_mark: All tests successful. No failed tests found.
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## main #6132 +/- ##
============================================
- Coverage 59.12% 59.12% -0.01%
+ Complexity 3208 3207 -1
============================================
Files 1132 1132
Lines 43681 43681
Branches 4734 4734
============================================
- Hits 25828 25825 -3
- Misses 16417 16420 +3
Partials 1436 1436
```
| [Flag](https://app.codecov.io/gh/apache/texera/pull/6132/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | | *Carryforward flag |
|---|---|---|---|
| [access-control-service](https://app.codecov.io/gh/apache/texera/pull/6132/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `70.00% <ø> (ø)` | | Carriedforward from [3a12e62](https://app.codecov.io/gh/apache/texera/commit/3a12e6214c5f45477042374ae3a4a569835d1fda?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [agent-service](https://app.codecov.io/gh/apache/texera/pull/6132/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `44.59% <ø> (ø)` | | Carriedforward from [3a12e62](https://app.codecov.io/gh/apache/texera/commit/3a12e6214c5f45477042374ae3a4a569835d1fda?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [amber](https://app.codecov.io/gh/apache/texera/pull/6132/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `63.14% <ø> (-0.02%)` | :arrow_down: | |
| [computing-unit-managing-service](https://app.codecov.io/gh/apache/texera/pull/6132/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <ø> (ø)` | | Carriedforward from [3a12e62](https://app.codecov.io/gh/apache/texera/commit/3a12e6214c5f45477042374ae3a4a569835d1fda?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [config-service](https://app.codecov.io/gh/apache/texera/pull/6132/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `52.30% <ø> (ø)` | | Carriedforward from [3a12e62](https://app.codecov.io/gh/apache/texera/commit/3a12e6214c5f45477042374ae3a4a569835d1fda?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [file-service](https://app.codecov.io/gh/apache/texera/pull/6132/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `62.81% <ø> (ø)` | | Carriedforward from [3a12e62](https://app.codecov.io/gh/apache/texera/commit/3a12e6214c5f45477042374ae3a4a569835d1fda?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [frontend](https://app.codecov.io/gh/apache/texera/pull/6132/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `51.31% <ø> (ø)` | | Carriedforward from [3a12e62](https://app.codecov.io/gh/apache/texera/commit/3a12e6214c5f45477042374ae3a4a569835d1fda?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [notebook-migration-service](https://app.codecov.io/gh/apache/texera/pull/6132/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `78.57% <ø> (ø)` | | Carriedforward from [3a12e62](https://app.codecov.io/gh/apache/texera/commit/3a12e6214c5f45477042374ae3a4a569835d1fda?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [pyamber](https://app.codecov.io/gh/apache/texera/pull/6132/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `91.19% <ø> (ø)` | | Carriedforward from [3a12e62](https://app.codecov.io/gh/apache/texera/commit/3a12e6214c5f45477042374ae3a4a569835d1fda?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [workflow-compiling-service](https://app.codecov.io/gh/apache/texera/pull/6132/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `55.14% <ø> (ø)` | | Carriedforward from [3a12e62](https://app.codecov.io/gh/apache/texera/commit/3a12e6214c5f45477042374ae3a4a569835d1fda?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
*This pull request uses carry forward flags. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to find out more.
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/texera/pull/6132?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(amber): add X-shaped multi-region workflow e2e integration test [texera]
Posted by "aglinxinyuan (via GitHub)" <gi...@apache.org>.
aglinxinyuan commented on PR #6132:
URL: https://github.com/apache/texera/pull/6132#issuecomment-4885226216
Thanks @Yicong-Huang — addressed all three:
1. **Native Sort** — swapped the hand-rolled Python sort UDF for the native `SortOpDesc` (pandas `sort_values`).
2. **Operator coverage** — reworked the X into one representative per family: CSV scan, hash join, aggregate, native Sort, union, projection, Python UDF. Join-build / aggregate / sort are all blocking, so the plan still spans several regions crossed by the reader-thread path.
3. **PR description** — rewritten to match the code (the old diagram still described the pre-rework Python-source version).
Assertions now: terminal A (join→aggregate→sort) → 7 per-region COUNTs, sorted by `Region`, summing to 100; terminal B (union→projection→UDF) → 200 rows narrowed to `{Region, Order ID}`. Compiles + scalafmt + scalafix clean locally; the run itself is covered by the `amber-integration` job.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(amber): add X-shaped multi-region workflow e2e integration test [texera]
Posted by "aglinxinyuan (via GitHub)" <gi...@apache.org>.
aglinxinyuan commented on code in PR #6132:
URL: https://github.com/apache/texera/pull/6132#discussion_r3524231965
##########
amber/src/test/integration/org/apache/texera/amber/engine/e2e/MultiRegionWorkflowIntegrationSpec.scala:
##########
@@ -0,0 +1,286 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.amber.engine.e2e
+
+import com.twitter.util.{Await, Duration, Promise, Return}
+import com.typesafe.scalalogging.Logger
+import org.apache.pekko.actor.{ActorSystem, Props}
+import org.apache.pekko.testkit.{ImplicitSender, TestKit}
+import org.apache.pekko.util.Timeout
+import org.apache.texera.amber.clustering.SingleNodeListener
+import org.apache.texera.amber.core.workflow.PortIdentity
+import org.apache.texera.amber.engine.architecture.controller.{
+ ControllerConfig,
+ ExecutionStateUpdate
+}
+import org.apache.texera.amber.engine.architecture.rpc.controlcommands.EmptyRequest
+import org.apache.texera.amber.engine.architecture.rpc.controlreturns.WorkflowAggregatedState.COMPLETED
+import org.apache.texera.amber.engine.architecture.scheduling.CostBasedScheduleGenerator
+import org.apache.texera.amber.engine.common.AmberRuntime
+import org.apache.texera.amber.engine.common.client.AmberClient
+import org.apache.texera.amber.engine.common.virtualidentity.util.CONTROLLER
+import org.apache.texera.amber.engine.e2e.TestUtils.{
+ buildWorkflow,
+ cleanupWorkflowExecutionData,
+ initiateTexeraDBForTestCases,
+ runWorkflowAndReadTerminalResults,
+ setUpWorkflowExecutionData
+}
+import org.apache.texera.amber.operator.TestOperators
+import org.apache.texera.amber.operator.source.scan.text.TextInputSourceOpDesc
+import org.apache.texera.amber.operator.union.UnionOpDesc
+import org.apache.texera.amber.tags.IntegrationTest
+import org.apache.texera.workflow.LogicalLink
+import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach, Outcome, Retries}
+import org.scalatest.flatspec.AnyFlatSpecLike
+
+import scala.concurrent.duration._
+
+/**
+ * End-to-end coverage for a workflow that executes across MULTIPLE regions.
+ *
+ * Cross-region data used to be delivered by dedicated cache-source operators;
+ * #3425 replaced them with input-port materialization reader threads. Before
+ * removing the now-dead cache-source plumbing, this spec adds the missing
+ * end-to-end coverage for that multi-region path: until now it was exercised
+ * only by scheduling unit tests (region counting on a physical plan), so a
+ * regression in the materialization read/write path could pass CI unnoticed.
+ *
+ * This spec drives a big "X"-shaped workflow to guard that path:
+ *
+ * {{{
+ * csvLeft ─┬─▶ join.build (port 0) ─┐
+ * │ join ─▶ pythonSort ─▶ (terminal)
+ * csvRight ─┼─▶ join.probe (port 1) ─┘
+ * │
+ * csvLeft ─┼─▶ union ───────────────────────────────────▶ (terminal)
+ * csvRight ─┘ (two links fan into union's single port)
+ * }}}
+ *
+ * The hash join's probe-depends-on-build ordering forces the build input to be
+ * materialized, cutting the plan into >=2 regions whose boundary is crossed by
+ * the reader-thread path. A Python UDF (sort) makes it a real end-to-end run
+ * with worker subprocesses, so it is class-level `@IntegrationTest` tagged and
+ * routed to the `amber-integration` CI job (which provisions Python deps); the
+ * lighter `amber` job excludes this tag.
+ */
+@IntegrationTest
+class MultiRegionWorkflowIntegrationSpec
+ extends TestKit(ActorSystem("MultiRegionWorkflowIntegrationSpec", AmberRuntime.pekkoConfig))
+ with ImplicitSender
+ with AnyFlatSpecLike
+ with BeforeAndAfterAll
+ with BeforeAndAfterEach
+ with Retries {
+
+ /**
+ * Retry each test once if it fails. Mirrors the other e2e integration specs:
+ * in CI there is a small chance the run does not observe "COMPLETED", so a
+ * single retry stabilizes the suite until that root cause is fixed.
+ */
+ override def withFixture(test: NoArgTest): Outcome =
+ withRetry { super.withFixture(test) }
+
+ implicit val timeout: Timeout = Timeout(5.seconds)
+
+ private val logger = Logger("MultiRegionWorkflowIntegrationSpecLogger")
+ private val specId = 5
+ private val ctx = TestUtils.workflowContext(specId)
+
+ // country_sales_small.csv has 100 rows with a unique "Order ID" per row.
+ private val sourceRowCount = 100
+
+ // Buffer every input tuple, then emit them ordered by "Order ID" once the
+ // input port is exhausted. This makes the UDF a genuine sort and forces a real
+ // Python worker to process data that crossed a region boundary.
+ private val sortByOrderIdCode =
+ """
+ |from pytexera import *
+ |
+ |class ProcessTupleOperator(UDFOperatorV2):
+ | def open(self) -> None:
+ | self.buffer = []
+ |
+ | @overrides
+ | def process_tuple(self, tuple_: Tuple, port: int) -> Iterator[Optional[TupleLike]]:
+ | self.buffer.append(tuple_)
+ | yield from []
+ |
+ | @overrides
+ | def on_finish(self, port: int) -> Iterator[Optional[TupleLike]]:
+ | for row in sorted(self.buffer, key=lambda t: t["Order ID"]):
+ | yield row
+ |""".stripMargin
+
+ override protected def beforeEach(): Unit = {
+ setUpWorkflowExecutionData(specId)
+ }
+
+ override protected def afterEach(): Unit = {
+ cleanupWorkflowExecutionData(specId)
+ }
+
+ override def beforeAll(): Unit = {
+ system.actorOf(Props[SingleNodeListener](), "cluster-info")
+ // These test cases access postgres in CI, but occasionally the jdbc driver cannot be found during CI run.
+ // Explicitly load the JDBC driver to avoid flaky CI failures.
+ Class.forName("org.postgresql.Driver")
+ initiateTexeraDBForTestCases()
+ warmupOnce()
+ }
+
+ override def afterAll(): Unit = {
+ TestKit.shutdownActorSystem(system)
+ }
+
+ /**
+ * Runs a TextInput -> Python UDF workflow once before the timed test so the
+ * Python worker cold-start is paid here, not inside the timed run. Capped and
+ * wrapped so warmup can never fail or hang the suite.
+ */
+ private def warmupOnce(): Unit = {
+ val warmupCap = Duration.fromSeconds(60)
+ setUpWorkflowExecutionData(specId)
+ var client: AmberClient = null
+ try {
+ val src = new TextInputSourceOpDesc()
+ src.textInput = "warmup"
+ val udf = TestOperators.pythonOpDesc()
+ val warmupCtx = TestUtils.workflowContext(specId)
+ val workflow = buildWorkflow(
+ List(src, udf),
+ List(
+ LogicalLink(
+ src.operatorIdentifier,
+ PortIdentity(),
+ udf.operatorIdentifier,
+ PortIdentity()
+ )
+ ),
+ warmupCtx
+ )
+ client = new AmberClient(
+ system,
+ workflow.context,
+ workflow.physicalPlan,
+ ControllerConfig.default,
+ _ => {}
+ )
+ val completion = Promise[Unit]()
+ client.registerCallback[ExecutionStateUpdate](evt => {
+ if (evt.state == COMPLETED) completion.updateIfEmpty(Return(()))
+ })
Review Comment:
Good catch — wired the error path (client callback + a FatalError callback) to fail the completion Promise immediately, so a broken warmup returns right away instead of blocking the full cap. Matches TestUtils.runWorkflowAndReadResults. Fixed in 70eb480.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(amber): add X-shaped multi-region workflow e2e integration test [texera]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #6132:
URL: https://github.com/apache/texera/pull/6132#issuecomment-4884665468
<!-- texera-benchmarks-comment -->
## ⚠️ Benchmark changes need a look
🟢 0 better · 🔴 10 worse · ⚪ 5 noise (<±5%) · 0 without baseline
> Compared against main `3a12e62` benchmarked on **this same runner**, so the delta is largely free of cross-runner hardware noise. The "7d avg" column still reflects the gh-pages dashboard. Treat <±5% as noise unless repeated.
[Dashboard](https://apache.github.io/texera/dev/bench/) · [Run](https://github.com/apache/texera/actions/runs/28727730657)
| | config | throughput | MB/s | latency | max Δ latest / 7d |
|---|---|---:|---:|---:|---:|
| 🔴 | bs=10 sw=10 sl=64 | 345 | 0.211 | 28,507/38,496/38,496 us | 🔴 +13.0% / 🔴 +154.2% |
| 🔴 | bs=100 sw=10 sl=64 | 729 | 0.445 | 133,535/173,682/173,682 us | 🔴 +13.8% / 🔴 +60.9% |
| ⚪ | bs=1000 sw=10 sl=64 | 880 | 0.537 | 1,137,113/1,149,745/1,149,745 us | ⚪ within ±5% / 🔴 +14.9% |
<details><summary>Baseline details</summary>
Latest main `3a12e62` from same runner
| config | metric | PR | latest main | 7d avg | Δ latest | Δ 7d |
|---|---|---:|---:|---:|---:|---:|
| bs=10 sw=10 sl=64 | throughput | 345 tuples/sec | 394 tuples/sec | 776.36 tuples/sec | -12.4% | -55.6% |
| bs=10 sw=10 sl=64 | MB/s | 0.211 MB/s | 0.24 MB/s | 0.474 MB/s | -12.1% | -55.5% |
| bs=10 sw=10 sl=64 | p50 | 28,507 us | 25,296 us | 12,636 us | +12.7% | +125.6% |
| bs=10 sw=10 sl=64 | p95 | 38,496 us | 34,065 us | 15,143 us | +13.0% | +154.2% |
| bs=10 sw=10 sl=64 | p99 | 38,496 us | 34,065 us | 18,954 us | +13.0% | +103.1% |
| bs=100 sw=10 sl=64 | throughput | 729 tuples/sec | 814 tuples/sec | 985.33 tuples/sec | -10.4% | -26.0% |
| bs=100 sw=10 sl=64 | MB/s | 0.445 MB/s | 0.497 MB/s | 0.601 MB/s | -10.5% | -26.0% |
| bs=100 sw=10 sl=64 | p50 | 133,535 us | 120,629 us | 101,671 us | +10.7% | +31.3% |
| bs=100 sw=10 sl=64 | p95 | 173,682 us | 152,681 us | 107,939 us | +13.8% | +60.9% |
| bs=100 sw=10 sl=64 | p99 | 173,682 us | 152,681 us | 113,798 us | +13.8% | +52.6% |
| bs=1000 sw=10 sl=64 | throughput | 880 tuples/sec | 905 tuples/sec | 1,016 tuples/sec | -2.8% | -13.4% |
| bs=1000 sw=10 sl=64 | MB/s | 0.537 MB/s | 0.552 MB/s | 0.62 MB/s | -2.7% | -13.4% |
| bs=1000 sw=10 sl=64 | p50 | 1,137,113 us | 1,107,224 us | 989,693 us | +2.7% | +14.9% |
| bs=1000 sw=10 sl=64 | p95 | 1,149,745 us | 1,163,101 us | 1,028,327 us | -1.1% | +11.8% |
| bs=1000 sw=10 sl=64 | p99 | 1,149,745 us | 1,163,101 us | 1,059,969 us | -1.1% | +8.5% |
</details>
<details><summary>Raw CSV</summary>
```csv
config_idx,batch_size,schema_width,string_len,num_batches,total_ms,total_tuples,total_bytes,tuples_per_sec,mb_per_sec,lat_p50_us,lat_p95_us,lat_p99_us
0,10,10,64,20,579.01,200,128000,345,0.211,28507.40,38495.77,38495.77
1,100,10,64,20,2744.77,2000,1280000,729,0.445,133534.52,173681.95,173681.95
2,1000,10,64,20,22732.36,20000,12800000,880,0.537,1137112.58,1149744.71,1149744.71
```
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(amber): add X-shaped multi-region workflow e2e integration test [texera]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #6132:
URL: https://github.com/apache/texera/pull/6132#discussion_r3524221561
##########
amber/src/test/integration/org/apache/texera/amber/engine/e2e/MultiRegionWorkflowIntegrationSpec.scala:
##########
@@ -0,0 +1,286 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.amber.engine.e2e
+
+import com.twitter.util.{Await, Duration, Promise, Return}
+import com.typesafe.scalalogging.Logger
+import org.apache.pekko.actor.{ActorSystem, Props}
+import org.apache.pekko.testkit.{ImplicitSender, TestKit}
+import org.apache.pekko.util.Timeout
+import org.apache.texera.amber.clustering.SingleNodeListener
+import org.apache.texera.amber.core.workflow.PortIdentity
+import org.apache.texera.amber.engine.architecture.controller.{
+ ControllerConfig,
+ ExecutionStateUpdate
+}
+import org.apache.texera.amber.engine.architecture.rpc.controlcommands.EmptyRequest
+import org.apache.texera.amber.engine.architecture.rpc.controlreturns.WorkflowAggregatedState.COMPLETED
+import org.apache.texera.amber.engine.architecture.scheduling.CostBasedScheduleGenerator
+import org.apache.texera.amber.engine.common.AmberRuntime
+import org.apache.texera.amber.engine.common.client.AmberClient
+import org.apache.texera.amber.engine.common.virtualidentity.util.CONTROLLER
+import org.apache.texera.amber.engine.e2e.TestUtils.{
+ buildWorkflow,
+ cleanupWorkflowExecutionData,
+ initiateTexeraDBForTestCases,
+ runWorkflowAndReadTerminalResults,
+ setUpWorkflowExecutionData
+}
+import org.apache.texera.amber.operator.TestOperators
+import org.apache.texera.amber.operator.source.scan.text.TextInputSourceOpDesc
+import org.apache.texera.amber.operator.union.UnionOpDesc
+import org.apache.texera.amber.tags.IntegrationTest
+import org.apache.texera.workflow.LogicalLink
+import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach, Outcome, Retries}
+import org.scalatest.flatspec.AnyFlatSpecLike
+
+import scala.concurrent.duration._
+
+/**
+ * End-to-end coverage for a workflow that executes across MULTIPLE regions.
+ *
+ * Cross-region data used to be delivered by dedicated cache-source operators;
+ * #3425 replaced them with input-port materialization reader threads. Before
+ * removing the now-dead cache-source plumbing, this spec adds the missing
+ * end-to-end coverage for that multi-region path: until now it was exercised
+ * only by scheduling unit tests (region counting on a physical plan), so a
+ * regression in the materialization read/write path could pass CI unnoticed.
+ *
+ * This spec drives a big "X"-shaped workflow to guard that path:
+ *
+ * {{{
+ * csvLeft ─┬─▶ join.build (port 0) ─┐
+ * │ join ─▶ pythonSort ─▶ (terminal)
+ * csvRight ─┼─▶ join.probe (port 1) ─┘
+ * │
+ * csvLeft ─┼─▶ union ───────────────────────────────────▶ (terminal)
+ * csvRight ─┘ (two links fan into union's single port)
+ * }}}
+ *
+ * The hash join's probe-depends-on-build ordering forces the build input to be
+ * materialized, cutting the plan into >=2 regions whose boundary is crossed by
+ * the reader-thread path. A Python UDF (sort) makes it a real end-to-end run
+ * with worker subprocesses, so it is class-level `@IntegrationTest` tagged and
+ * routed to the `amber-integration` CI job (which provisions Python deps); the
+ * lighter `amber` job excludes this tag.
+ */
+@IntegrationTest
+class MultiRegionWorkflowIntegrationSpec
+ extends TestKit(ActorSystem("MultiRegionWorkflowIntegrationSpec", AmberRuntime.pekkoConfig))
+ with ImplicitSender
+ with AnyFlatSpecLike
+ with BeforeAndAfterAll
+ with BeforeAndAfterEach
+ with Retries {
+
+ /**
+ * Retry each test once if it fails. Mirrors the other e2e integration specs:
+ * in CI there is a small chance the run does not observe "COMPLETED", so a
+ * single retry stabilizes the suite until that root cause is fixed.
+ */
+ override def withFixture(test: NoArgTest): Outcome =
+ withRetry { super.withFixture(test) }
+
+ implicit val timeout: Timeout = Timeout(5.seconds)
+
+ private val logger = Logger("MultiRegionWorkflowIntegrationSpecLogger")
+ private val specId = 5
+ private val ctx = TestUtils.workflowContext(specId)
+
+ // country_sales_small.csv has 100 rows with a unique "Order ID" per row.
+ private val sourceRowCount = 100
+
+ // Buffer every input tuple, then emit them ordered by "Order ID" once the
+ // input port is exhausted. This makes the UDF a genuine sort and forces a real
+ // Python worker to process data that crossed a region boundary.
+ private val sortByOrderIdCode =
+ """
+ |from pytexera import *
+ |
+ |class ProcessTupleOperator(UDFOperatorV2):
+ | def open(self) -> None:
+ | self.buffer = []
+ |
+ | @overrides
+ | def process_tuple(self, tuple_: Tuple, port: int) -> Iterator[Optional[TupleLike]]:
+ | self.buffer.append(tuple_)
+ | yield from []
+ |
+ | @overrides
+ | def on_finish(self, port: int) -> Iterator[Optional[TupleLike]]:
+ | for row in sorted(self.buffer, key=lambda t: t["Order ID"]):
+ | yield row
+ |""".stripMargin
+
+ override protected def beforeEach(): Unit = {
+ setUpWorkflowExecutionData(specId)
+ }
+
+ override protected def afterEach(): Unit = {
+ cleanupWorkflowExecutionData(specId)
+ }
+
+ override def beforeAll(): Unit = {
+ system.actorOf(Props[SingleNodeListener](), "cluster-info")
+ // These test cases access postgres in CI, but occasionally the jdbc driver cannot be found during CI run.
+ // Explicitly load the JDBC driver to avoid flaky CI failures.
+ Class.forName("org.postgresql.Driver")
+ initiateTexeraDBForTestCases()
+ warmupOnce()
+ }
+
+ override def afterAll(): Unit = {
+ TestKit.shutdownActorSystem(system)
+ }
+
+ /**
+ * Runs a TextInput -> Python UDF workflow once before the timed test so the
+ * Python worker cold-start is paid here, not inside the timed run. Capped and
+ * wrapped so warmup can never fail or hang the suite.
+ */
+ private def warmupOnce(): Unit = {
+ val warmupCap = Duration.fromSeconds(60)
+ setUpWorkflowExecutionData(specId)
+ var client: AmberClient = null
+ try {
+ val src = new TextInputSourceOpDesc()
+ src.textInput = "warmup"
+ val udf = TestOperators.pythonOpDesc()
+ val warmupCtx = TestUtils.workflowContext(specId)
+ val workflow = buildWorkflow(
+ List(src, udf),
+ List(
+ LogicalLink(
+ src.operatorIdentifier,
+ PortIdentity(),
+ udf.operatorIdentifier,
+ PortIdentity()
+ )
+ ),
+ warmupCtx
+ )
+ client = new AmberClient(
+ system,
+ workflow.context,
+ workflow.physicalPlan,
+ ControllerConfig.default,
+ _ => {}
+ )
+ val completion = Promise[Unit]()
+ client.registerCallback[ExecutionStateUpdate](evt => {
+ if (evt.state == COMPLETED) completion.updateIfEmpty(Return(()))
+ })
Review Comment:
In warmupOnce, the AmberClient error callback is `_ => {}` and the completion Promise is only completed on COMPLETED. If the warmup workflow hits a FatalError (or the client reports an error), the suite can wait the full warmupCap (60s) before timing out, which slows CI and contradicts the intent that warmup should never hang the suite. Consider wiring the error path to fail the Promise immediately (similar to TestUtils.runWorkflowAndReadResults).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(amber): add X-shaped multi-region workflow e2e integration test [texera]
Posted by "aglinxinyuan (via GitHub)" <gi...@apache.org>.
aglinxinyuan commented on code in PR #6132:
URL: https://github.com/apache/texera/pull/6132#discussion_r3524376840
##########
amber/src/test/integration/org/apache/texera/amber/engine/e2e/MultiRegionWorkflowIntegrationSpec.scala:
##########
@@ -0,0 +1,286 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.amber.engine.e2e
+
+import com.twitter.util.{Await, Duration, Promise, Return}
+import com.typesafe.scalalogging.Logger
+import org.apache.pekko.actor.{ActorSystem, Props}
+import org.apache.pekko.testkit.{ImplicitSender, TestKit}
+import org.apache.pekko.util.Timeout
+import org.apache.texera.amber.clustering.SingleNodeListener
+import org.apache.texera.amber.core.workflow.PortIdentity
+import org.apache.texera.amber.engine.architecture.controller.{
+ ControllerConfig,
+ ExecutionStateUpdate
+}
+import org.apache.texera.amber.engine.architecture.rpc.controlcommands.EmptyRequest
+import org.apache.texera.amber.engine.architecture.rpc.controlreturns.WorkflowAggregatedState.COMPLETED
+import org.apache.texera.amber.engine.architecture.scheduling.CostBasedScheduleGenerator
+import org.apache.texera.amber.engine.common.AmberRuntime
+import org.apache.texera.amber.engine.common.client.AmberClient
+import org.apache.texera.amber.engine.common.virtualidentity.util.CONTROLLER
+import org.apache.texera.amber.engine.e2e.TestUtils.{
+ buildWorkflow,
+ cleanupWorkflowExecutionData,
+ initiateTexeraDBForTestCases,
+ runWorkflowAndReadTerminalResults,
+ setUpWorkflowExecutionData
+}
+import org.apache.texera.amber.operator.TestOperators
+import org.apache.texera.amber.operator.source.scan.text.TextInputSourceOpDesc
+import org.apache.texera.amber.operator.union.UnionOpDesc
+import org.apache.texera.amber.tags.IntegrationTest
+import org.apache.texera.workflow.LogicalLink
+import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach, Outcome, Retries}
+import org.scalatest.flatspec.AnyFlatSpecLike
+
+import scala.concurrent.duration._
+
+/**
+ * End-to-end coverage for a workflow that executes across MULTIPLE regions.
+ *
+ * Cross-region data used to be delivered by dedicated cache-source operators;
+ * #3425 replaced them with input-port materialization reader threads. Before
+ * removing the now-dead cache-source plumbing, this spec adds the missing
+ * end-to-end coverage for that multi-region path: until now it was exercised
+ * only by scheduling unit tests (region counting on a physical plan), so a
+ * regression in the materialization read/write path could pass CI unnoticed.
+ *
+ * This spec drives a big "X"-shaped workflow to guard that path:
+ *
+ * {{{
+ * csvLeft ─┬─▶ join.build (port 0) ─┐
+ * │ join ─▶ pythonSort ─▶ (terminal)
+ * csvRight ─┼─▶ join.probe (port 1) ─┘
+ * │
+ * csvLeft ─┼─▶ union ───────────────────────────────────▶ (terminal)
+ * csvRight ─┘ (two links fan into union's single port)
+ * }}}
+ *
+ * The hash join's probe-depends-on-build ordering forces the build input to be
+ * materialized, cutting the plan into >=2 regions whose boundary is crossed by
+ * the reader-thread path. A Python UDF (sort) makes it a real end-to-end run
+ * with worker subprocesses, so it is class-level `@IntegrationTest` tagged and
+ * routed to the `amber-integration` CI job (which provisions Python deps); the
+ * lighter `amber` job excludes this tag.
+ */
+@IntegrationTest
+class MultiRegionWorkflowIntegrationSpec
+ extends TestKit(ActorSystem("MultiRegionWorkflowIntegrationSpec", AmberRuntime.pekkoConfig))
+ with ImplicitSender
+ with AnyFlatSpecLike
+ with BeforeAndAfterAll
+ with BeforeAndAfterEach
+ with Retries {
+
+ /**
+ * Retry each test once if it fails. Mirrors the other e2e integration specs:
+ * in CI there is a small chance the run does not observe "COMPLETED", so a
+ * single retry stabilizes the suite until that root cause is fixed.
+ */
+ override def withFixture(test: NoArgTest): Outcome =
+ withRetry { super.withFixture(test) }
+
+ implicit val timeout: Timeout = Timeout(5.seconds)
+
+ private val logger = Logger("MultiRegionWorkflowIntegrationSpecLogger")
+ private val specId = 5
+ private val ctx = TestUtils.workflowContext(specId)
+
+ // country_sales_small.csv has 100 rows with a unique "Order ID" per row.
+ private val sourceRowCount = 100
+
+ // Buffer every input tuple, then emit them ordered by "Order ID" once the
+ // input port is exhausted. This makes the UDF a genuine sort and forces a real
+ // Python worker to process data that crossed a region boundary.
+ private val sortByOrderIdCode =
+ """
+ |from pytexera import *
+ |
+ |class ProcessTupleOperator(UDFOperatorV2):
+ | def open(self) -> None:
+ | self.buffer = []
+ |
+ | @overrides
+ | def process_tuple(self, tuple_: Tuple, port: int) -> Iterator[Optional[TupleLike]]:
+ | self.buffer.append(tuple_)
+ | yield from []
+ |
+ | @overrides
+ | def on_finish(self, port: int) -> Iterator[Optional[TupleLike]]:
+ | for row in sorted(self.buffer, key=lambda t: t["Order ID"]):
+ | yield row
+ |""".stripMargin
+
+ override protected def beforeEach(): Unit = {
+ setUpWorkflowExecutionData(specId)
+ }
+
+ override protected def afterEach(): Unit = {
+ cleanupWorkflowExecutionData(specId)
+ }
+
+ override def beforeAll(): Unit = {
+ system.actorOf(Props[SingleNodeListener](), "cluster-info")
+ // These test cases access postgres in CI, but occasionally the jdbc driver cannot be found during CI run.
+ // Explicitly load the JDBC driver to avoid flaky CI failures.
+ Class.forName("org.postgresql.Driver")
+ initiateTexeraDBForTestCases()
+ warmupOnce()
+ }
+
+ override def afterAll(): Unit = {
+ TestKit.shutdownActorSystem(system)
+ }
+
+ /**
+ * Runs a TextInput -> Python UDF workflow once before the timed test so the
+ * Python worker cold-start is paid here, not inside the timed run. Capped and
+ * wrapped so warmup can never fail or hang the suite.
+ */
+ private def warmupOnce(): Unit = {
+ val warmupCap = Duration.fromSeconds(60)
+ setUpWorkflowExecutionData(specId)
+ var client: AmberClient = null
+ try {
+ val src = new TextInputSourceOpDesc()
+ src.textInput = "warmup"
+ val udf = TestOperators.pythonOpDesc()
+ val warmupCtx = TestUtils.workflowContext(specId)
+ val workflow = buildWorkflow(
+ List(src, udf),
+ List(
+ LogicalLink(
+ src.operatorIdentifier,
+ PortIdentity(),
+ udf.operatorIdentifier,
+ PortIdentity()
+ )
+ ),
+ warmupCtx
+ )
+ client = new AmberClient(
+ system,
+ workflow.context,
+ workflow.physicalPlan,
+ ControllerConfig.default,
+ _ => {}
+ )
+ val completion = Promise[Unit]()
+ client.registerCallback[ExecutionStateUpdate](evt => {
+ if (evt.state == COMPLETED) completion.updateIfEmpty(Return(()))
+ })
Review Comment:
This is already handled on the current head: the AmberClient error callback is `e => completion.updateIfEmpty(Throw(e))` and a `FatalError` callback also fails the completion Promise, so a broken warmup returns immediately instead of waiting out the 60s cap (the surrounding try/catch then logs and lets tests run cold). Resolving.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(amber): add X-shaped multi-region workflow e2e integration test [texera]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #6132:
URL: https://github.com/apache/texera/pull/6132#issuecomment-4884646922
<!-- texera-reviewer-suggestion -->
### Automated Reviewer Suggestions
Based on the `git blame` history of the changed files, we recommend the following reviewers:
- No candidates found from `git blame` history.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(amber): add X-shaped multi-region workflow e2e integration test [texera]
Posted by "aglinxinyuan (via GitHub)" <gi...@apache.org>.
aglinxinyuan commented on code in PR #6132:
URL: https://github.com/apache/texera/pull/6132#discussion_r3524392229
##########
amber/src/test/integration/org/apache/texera/amber/engine/e2e/MultiRegionWorkflowIntegrationSpec.scala:
##########
@@ -0,0 +1,291 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.amber.engine.e2e
+
+import com.twitter.util.{Await, Duration, Promise, Return, Throw}
+import com.typesafe.scalalogging.Logger
+import org.apache.pekko.actor.{ActorSystem, Props}
+import org.apache.pekko.testkit.{ImplicitSender, TestKit}
+import org.apache.pekko.util.Timeout
+import org.apache.texera.amber.clustering.SingleNodeListener
+import org.apache.texera.amber.core.workflow.PortIdentity
+import org.apache.texera.amber.engine.architecture.controller.{
+ ControllerConfig,
+ ExecutionStateUpdate,
+ FatalError
+}
+import org.apache.texera.amber.engine.architecture.rpc.controlcommands.EmptyRequest
+import org.apache.texera.amber.engine.architecture.rpc.controlreturns.WorkflowAggregatedState.COMPLETED
+import org.apache.texera.amber.engine.architecture.scheduling.CostBasedScheduleGenerator
+import org.apache.texera.amber.engine.common.AmberRuntime
+import org.apache.texera.amber.engine.common.client.AmberClient
+import org.apache.texera.amber.engine.common.virtualidentity.util.CONTROLLER
+import org.apache.texera.amber.engine.e2e.TestUtils.{
+ buildWorkflow,
+ cleanupWorkflowExecutionData,
+ initiateTexeraDBForTestCases,
+ runWorkflowAndReadTerminalResults,
+ setUpWorkflowExecutionData
+}
+import org.apache.texera.amber.operator.TestOperators
+import org.apache.texera.amber.operator.source.scan.text.TextInputSourceOpDesc
+import org.apache.texera.amber.operator.union.UnionOpDesc
+import org.apache.texera.amber.tags.IntegrationTest
+import org.apache.texera.workflow.LogicalLink
+import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach, Outcome, Retries}
+import org.scalatest.flatspec.AnyFlatSpecLike
+
+import scala.concurrent.duration._
+
+/**
+ * End-to-end coverage for a workflow that executes across MULTIPLE regions.
+ *
+ * Cross-region data used to be delivered by dedicated cache-source operators;
+ * #3425 replaced them with input-port materialization reader threads. Before
+ * removing the now-dead cache-source plumbing, this spec adds the missing
+ * end-to-end coverage for that multi-region path: until now it was exercised
+ * only by scheduling unit tests (region counting on a physical plan), so a
+ * regression in the materialization read/write path could pass CI unnoticed.
+ *
+ * This spec drives a big "X"-shaped workflow to guard that path:
+ *
+ * {{{
+ * csvLeft ─┬─▶ join.build (port 0) ─┐
+ * │ join ─▶ pythonSort ─▶ (terminal)
+ * csvRight ─┼─▶ join.probe (port 1) ─┘
+ * │
+ * csvLeft ─┼─▶ union ───────────────────────────────────▶ (terminal)
+ * csvRight ─┘ (two links fan into union's single port)
+ * }}}
+ *
+ * The hash join's probe-depends-on-build ordering forces the build input to be
+ * materialized, cutting the plan into >=2 regions whose boundary is crossed by
+ * the reader-thread path. A Python UDF (sort) makes it a real end-to-end run
+ * with worker subprocesses, so it is class-level `@IntegrationTest` tagged and
+ * routed to the `amber-integration` CI job (which provisions Python deps); the
+ * lighter `amber` job excludes this tag.
+ */
+@IntegrationTest
+class MultiRegionWorkflowIntegrationSpec
+ extends TestKit(ActorSystem("MultiRegionWorkflowIntegrationSpec", AmberRuntime.pekkoConfig))
+ with ImplicitSender
+ with AnyFlatSpecLike
+ with BeforeAndAfterAll
+ with BeforeAndAfterEach
+ with Retries {
+
+ /**
+ * Retry each test once if it fails. Mirrors the other e2e integration specs:
+ * in CI there is a small chance the run does not observe "COMPLETED", so a
+ * single retry stabilizes the suite until that root cause is fixed.
+ */
+ override def withFixture(test: NoArgTest): Outcome =
+ withRetry { super.withFixture(test) }
+
+ implicit val timeout: Timeout = Timeout(5.seconds)
+
+ private val logger = Logger("MultiRegionWorkflowIntegrationSpecLogger")
+ private val specId = 5
+ private val ctx = TestUtils.workflowContext(specId)
+
+ // country_sales_small.csv has 100 rows with a unique "Order ID" per row.
+ private val sourceRowCount = 100
+
+ // Buffer every input tuple, then emit them ordered by "Order ID" once the
+ // input port is exhausted. This makes the UDF a genuine sort and forces a real
+ // Python worker to process data that crossed a region boundary.
+ private val sortByOrderIdCode =
+ """
+ |from pytexera import *
+ |
+ |class ProcessTupleOperator(UDFOperatorV2):
+ | def open(self) -> None:
+ | self.buffer = []
+ |
+ | @overrides
+ | def process_tuple(self, tuple_: Tuple, port: int) -> Iterator[Optional[TupleLike]]:
+ | self.buffer.append(tuple_)
+ | yield from []
+ |
+ | @overrides
+ | def on_finish(self, port: int) -> Iterator[Optional[TupleLike]]:
+ | for row in sorted(self.buffer, key=lambda t: t["Order ID"]):
+ | yield row
+ |""".stripMargin
+
+ override protected def beforeEach(): Unit = {
+ setUpWorkflowExecutionData(specId)
+ }
+
+ override protected def afterEach(): Unit = {
+ cleanupWorkflowExecutionData(specId)
+ }
+
+ override def beforeAll(): Unit = {
+ system.actorOf(Props[SingleNodeListener](), "cluster-info")
+ // These test cases access postgres in CI, but occasionally the jdbc driver cannot be found during CI run.
+ // Explicitly load the JDBC driver to avoid flaky CI failures.
+ Class.forName("org.postgresql.Driver")
+ initiateTexeraDBForTestCases()
+ warmupOnce()
+ }
+
+ override def afterAll(): Unit = {
+ TestKit.shutdownActorSystem(system)
+ }
+
+ /**
+ * Runs a TextInput -> Python UDF workflow once before the timed test so the
+ * Python worker cold-start is paid here, not inside the timed run. Capped and
+ * wrapped so warmup can never fail or hang the suite.
+ */
+ private def warmupOnce(): Unit = {
+ val warmupCap = Duration.fromSeconds(60)
+ setUpWorkflowExecutionData(specId)
+ var client: AmberClient = null
+ try {
+ val src = new TextInputSourceOpDesc()
+ src.textInput = "warmup"
+ val udf = TestOperators.pythonOpDesc()
+ val warmupCtx = TestUtils.workflowContext(specId)
+ val workflow = buildWorkflow(
+ List(src, udf),
+ List(
+ LogicalLink(
+ src.operatorIdentifier,
+ PortIdentity(),
+ udf.operatorIdentifier,
+ PortIdentity()
+ )
+ ),
+ warmupCtx
+ )
+ // Fail the completion Promise on any error (client-reported or FatalError)
+ // so a broken warmup returns immediately instead of blocking the whole
+ // warmupCap; the surrounding try/catch then logs and lets tests run cold.
+ val completion = Promise[Unit]()
+ client = new AmberClient(
+ system,
+ workflow.context,
+ workflow.physicalPlan,
+ ControllerConfig.default,
+ e => completion.updateIfEmpty(Throw(e))
+ )
+ client.registerCallback[FatalError](evt => completion.updateIfEmpty(Throw(evt.e)))
+ client.registerCallback[ExecutionStateUpdate](evt => {
+ if (evt.state == COMPLETED) completion.updateIfEmpty(Return(()))
+ })
+ Await.result(
+ client.controllerInterface.startWorkflow(EmptyRequest(), ()),
+ warmupCap
+ )
+ Await.result(completion, warmupCap)
+ } catch {
+ case e: Throwable =>
+ logger.warn(
+ s"warmup workflow did not finish within ${warmupCap}; tests will run cold and rely on Retries: ${e.getMessage}"
+ )
+ } finally {
+ if (client != null) {
+ try client.shutdown()
+ catch { case _: Throwable => () }
+ }
+ cleanupWorkflowExecutionData(specId)
+ }
+ }
+
+ "Engine" should "execute an X-shaped multi-region workflow (join + union + python) end-to-end" in {
+ val csvLeft = TestOperators.smallCsvScanOpDesc()
+ val csvRight = TestOperators.smallCsvScanOpDesc()
+ // Self-join on the unique "Order ID" key: each row matches exactly one row
+ // from the other source, so the join emits exactly `sourceRowCount` rows.
+ val join = TestOperators.joinOpDesc("Order ID", "Order ID")
+ val pythonSort = TestOperators.pythonOpDesc()
+ pythonSort.code = sortByOrderIdCode
+ val union = new UnionOpDesc()
+
+ val workflow = buildWorkflow(
+ List(csvLeft, csvRight, join, pythonSort, union),
+ List(
+ // Left arm of the X: the two sources feed the join (build/probe) ...
+ LogicalLink(
+ csvLeft.operatorIdentifier,
+ PortIdentity(),
+ join.operatorIdentifier,
+ PortIdentity()
+ ),
+ LogicalLink(
+ csvRight.operatorIdentifier,
+ PortIdentity(),
+ join.operatorIdentifier,
+ PortIdentity(1)
+ ),
+ LogicalLink(
+ join.operatorIdentifier,
+ PortIdentity(),
+ pythonSort.operatorIdentifier,
+ PortIdentity()
+ ),
+ // ... and also fan into the union's single input port (the crossing).
+ LogicalLink(
+ csvLeft.operatorIdentifier,
+ PortIdentity(),
+ union.operatorIdentifier,
+ PortIdentity()
+ ),
+ LogicalLink(
+ csvRight.operatorIdentifier,
+ PortIdentity(),
+ union.operatorIdentifier,
+ PortIdentity()
+ )
+ ),
+ ctx
+ )
+
+ // The join's build/probe dependency cuts the plan into multiple regions
+ // whose boundary is crossed by the input-port materialization reader path.
+ // Assert that with the same generator the controller uses at runtime.
+ val (schedule, _) = new CostBasedScheduleGenerator(
+ workflow.context,
+ workflow.physicalPlan,
+ CONTROLLER
+ ).generate()
+ assert(
+ schedule.getRegions.size >= 2,
+ s"expected the X-shaped workflow to span multiple regions, got ${schedule.getRegions.size}"
+ )
Review Comment:
Agreed — made it a black box. Dropped the hand-wired CostBasedScheduleGenerator call and the region-count assertion; the test now just builds a logical plan and runs it through the real compiler/controller/scheduler, asserting only on the materialized terminal outputs. Correct results are only possible if the cross-region delivery worked, and the region count itself is already covered by the scheduling unit tests, so nothing here duplicates (or can drift from) the real scheduling path. In 6616dad.
##########
amber/src/test/integration/org/apache/texera/amber/engine/e2e/MultiRegionWorkflowIntegrationSpec.scala:
##########
@@ -0,0 +1,286 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.amber.engine.e2e
+
+import com.twitter.util.{Await, Duration, Promise, Return}
+import com.typesafe.scalalogging.Logger
+import org.apache.pekko.actor.{ActorSystem, Props}
+import org.apache.pekko.testkit.{ImplicitSender, TestKit}
+import org.apache.pekko.util.Timeout
+import org.apache.texera.amber.clustering.SingleNodeListener
+import org.apache.texera.amber.core.workflow.PortIdentity
+import org.apache.texera.amber.engine.architecture.controller.{
+ ControllerConfig,
+ ExecutionStateUpdate
+}
+import org.apache.texera.amber.engine.architecture.rpc.controlcommands.EmptyRequest
+import org.apache.texera.amber.engine.architecture.rpc.controlreturns.WorkflowAggregatedState.COMPLETED
+import org.apache.texera.amber.engine.architecture.scheduling.CostBasedScheduleGenerator
+import org.apache.texera.amber.engine.common.AmberRuntime
+import org.apache.texera.amber.engine.common.client.AmberClient
+import org.apache.texera.amber.engine.common.virtualidentity.util.CONTROLLER
+import org.apache.texera.amber.engine.e2e.TestUtils.{
+ buildWorkflow,
+ cleanupWorkflowExecutionData,
+ initiateTexeraDBForTestCases,
+ runWorkflowAndReadTerminalResults,
+ setUpWorkflowExecutionData
+}
+import org.apache.texera.amber.operator.TestOperators
+import org.apache.texera.amber.operator.source.scan.text.TextInputSourceOpDesc
+import org.apache.texera.amber.operator.union.UnionOpDesc
+import org.apache.texera.amber.tags.IntegrationTest
+import org.apache.texera.workflow.LogicalLink
+import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach, Outcome, Retries}
+import org.scalatest.flatspec.AnyFlatSpecLike
+
+import scala.concurrent.duration._
+
+/**
+ * End-to-end coverage for a workflow that executes across MULTIPLE regions.
+ *
+ * Cross-region data used to be delivered by dedicated cache-source operators;
+ * #3425 replaced them with input-port materialization reader threads. Before
+ * removing the now-dead cache-source plumbing, this spec adds the missing
+ * end-to-end coverage for that multi-region path: until now it was exercised
+ * only by scheduling unit tests (region counting on a physical plan), so a
+ * regression in the materialization read/write path could pass CI unnoticed.
+ *
+ * This spec drives a big "X"-shaped workflow to guard that path:
+ *
+ * {{{
+ * csvLeft ─┬─▶ join.build (port 0) ─┐
+ * │ join ─▶ pythonSort ─▶ (terminal)
+ * csvRight ─┼─▶ join.probe (port 1) ─┘
+ * │
+ * csvLeft ─┼─▶ union ───────────────────────────────────▶ (terminal)
+ * csvRight ─┘ (two links fan into union's single port)
+ * }}}
+ *
+ * The hash join's probe-depends-on-build ordering forces the build input to be
+ * materialized, cutting the plan into >=2 regions whose boundary is crossed by
+ * the reader-thread path. A Python UDF (sort) makes it a real end-to-end run
+ * with worker subprocesses, so it is class-level `@IntegrationTest` tagged and
+ * routed to the `amber-integration` CI job (which provisions Python deps); the
+ * lighter `amber` job excludes this tag.
+ */
+@IntegrationTest
+class MultiRegionWorkflowIntegrationSpec
+ extends TestKit(ActorSystem("MultiRegionWorkflowIntegrationSpec", AmberRuntime.pekkoConfig))
+ with ImplicitSender
+ with AnyFlatSpecLike
+ with BeforeAndAfterAll
+ with BeforeAndAfterEach
+ with Retries {
+
+ /**
+ * Retry each test once if it fails. Mirrors the other e2e integration specs:
+ * in CI there is a small chance the run does not observe "COMPLETED", so a
+ * single retry stabilizes the suite until that root cause is fixed.
+ */
+ override def withFixture(test: NoArgTest): Outcome =
+ withRetry { super.withFixture(test) }
+
+ implicit val timeout: Timeout = Timeout(5.seconds)
+
+ private val logger = Logger("MultiRegionWorkflowIntegrationSpecLogger")
+ private val specId = 5
+ private val ctx = TestUtils.workflowContext(specId)
+
+ // country_sales_small.csv has 100 rows with a unique "Order ID" per row.
+ private val sourceRowCount = 100
+
+ // Buffer every input tuple, then emit them ordered by "Order ID" once the
+ // input port is exhausted. This makes the UDF a genuine sort and forces a real
+ // Python worker to process data that crossed a region boundary.
+ private val sortByOrderIdCode =
Review Comment:
Done — switched both sources to Python sources (TestOperators.pythonSourceOpDesc), so the whole pipeline now runs on real Python workers end to end. Kept the Python UDF for the sort per your note. I went with two Python sources rather than mixing in one because the union needs both inputs to share a schema. In 6616dad.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(amber): add X-shaped multi-region workflow e2e integration test [texera]
Posted by "aglinxinyuan (via GitHub)" <gi...@apache.org>.
aglinxinyuan merged PR #6132:
URL: https://github.com/apache/texera/pull/6132
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org