You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@texera.apache.org by "Yicong-Huang (via GitHub)" <gi...@apache.org> on 2026/05/03 02:49:56 UTC
[PR] test(amber): add unit tests for LinkConfig.toPartitioning [texera]
Yicong-Huang opened a new pull request, #4803:
URL: https://github.com/apache/texera/pull/4803
### What changes were proposed in this PR?
Adds scalatest coverage for `amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/config/LinkConfig.scala`. The `toPartitioning` partition-routing helper had no dedicated spec.
### Any related issues, documentation, discussions?
Closes #4801.
Two related bugs are pinned in the spec with explanatory comments and filed separately as Bug issues: (1) the `OneToOne` arm uses `List.zip` and silently truncates on unequal-length input — the same hazard as the parallel `ChannelConfig` issue (#4799). (2) The `BroadcastPartition` arm uses `zip` instead of the cross product, both diverging from the broadcast contract and from `ChannelConfig`'s broadcast routing in the same package.
### How was this PR tested?
```
sbt scalafmtCheckAll
sbt "WorkflowExecutionService/testOnly org.apache.texera.amber.engine.architecture.scheduling.config.LinkConfigSpec"
```
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (claude-opus-4-7)
--
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 unit tests for LinkConfig.toPartitioning [texera]
Posted by "aglinxinyuan (via GitHub)" <gi...@apache.org>.
aglinxinyuan merged PR #4803:
URL: https://github.com/apache/texera/pull/4803
--
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 unit tests for LinkConfig.toPartitioning [texera]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #4803:
URL: https://github.com/apache/texera/pull/4803#discussion_r3177548665
##########
amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/config/LinkConfigSpec.scala:
##########
@@ -0,0 +1,231 @@
+/*
+ * 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.architecture.scheduling.config
+
+import org.apache.texera.amber.core.virtualidentity.{ActorVirtualIdentity, ChannelIdentity}
+import org.apache.texera.amber.core.workflow.{
+ BroadcastPartition,
+ HashPartition,
+ OneToOnePartition,
+ RangePartition,
+ SinglePartition,
+ UnknownPartition
+}
+import org.apache.texera.amber.engine.architecture.sendsemantics.partitionings.{
+ BroadcastPartitioning,
+ HashBasedShufflePartitioning,
+ OneToOnePartitioning,
+ RangeBasedShufflePartitioning,
+ RoundRobinPartitioning
+}
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+class LinkConfigSpec extends AnyFlatSpec with Matchers {
+
+ private val w1 = ActorVirtualIdentity("w1")
+ private val w2 = ActorVirtualIdentity("w2")
+ private val w3 = ActorVirtualIdentity("w3")
+ private val u1 = ActorVirtualIdentity("u1")
+ private val u2 = ActorVirtualIdentity("u2")
+ private val u3 = ActorVirtualIdentity("u3")
+ private val batch = 64
+
+ private def endpoints(channels: Seq[ChannelIdentity]): Seq[(String, String)] =
+ channels.map(c => (c.fromWorkerId.name, c.toWorkerId.name))
+
+ // ----- HashPartition -----
+
+ "toPartitioning" should "produce a HashBasedShufflePartitioning with full cross product channels" in {
+ val out = LinkConfig.toPartitioning(
+ List(w1, w2),
+ List(u1, u2, u3),
+ HashPartition(List("k1", "k2")),
+ batch
+ )
+ out shouldBe a[HashBasedShufflePartitioning]
+ val hp = out.asInstanceOf[HashBasedShufflePartitioning]
+ hp.batchSize shouldBe batch
+ hp.hashAttributeNames shouldBe Seq("k1", "k2")
+ endpoints(hp.channels) shouldBe Seq(
+ ("w1", "u1"),
+ ("w1", "u2"),
+ ("w1", "u3"),
+ ("w2", "u1"),
+ ("w2", "u2"),
+ ("w2", "u3")
+ )
+ hp.channels.foreach(_.isControl shouldBe false)
+ }
+
+ // ----- RangePartition -----
+
+ "RangePartition" should "produce a RangeBasedShufflePartitioning carrying the range bounds and cross-product channels" in {
+ val out = LinkConfig.toPartitioning(
+ List(w1),
+ List(u1, u2),
+ RangePartition(List("k"), 0L, 100L),
+ batch
+ )
+ out shouldBe a[RangeBasedShufflePartitioning]
+ val rp = out.asInstanceOf[RangeBasedShufflePartitioning]
+ rp.batchSize shouldBe batch
+ rp.rangeAttributeNames shouldBe Seq("k")
+ rp.rangeMin shouldBe 0L
+ rp.rangeMax shouldBe 100L
+ endpoints(rp.channels) shouldBe Seq(("w1", "u1"), ("w1", "u2"))
+ }
+
+ // ----- SinglePartition -----
+
+ "SinglePartition" should "produce a OneToOnePartitioning with one channel per from-worker to the single to-worker" in {
+ val out = LinkConfig.toPartitioning(
+ List(w1, w2, w3),
+ List(u1),
+ SinglePartition(),
+ batch
+ )
+ out shouldBe a[OneToOnePartitioning]
+ val op = out.asInstanceOf[OneToOnePartitioning]
+ op.batchSize shouldBe batch
+ endpoints(op.channels) shouldBe Seq(("w1", "u1"), ("w2", "u1"), ("w3", "u1"))
+ }
+
+ it should "raise an AssertionError when more than one to-worker is supplied" in {
+ assertThrows[AssertionError] {
+ LinkConfig.toPartitioning(List(w1, w2), List(u1, u2), SinglePartition(), batch)
+ }
+ }
+
+ // ----- OneToOnePartition -----
+
+ "OneToOnePartition" should "produce a OneToOnePartitioning with zip pairing for equal-length inputs" in {
+ val out = LinkConfig.toPartitioning(
+ List(w1, w2, w3),
+ List(u1, u2, u3),
+ OneToOnePartition(),
+ batch
+ )
+ out shouldBe a[OneToOnePartitioning]
+ val op = out.asInstanceOf[OneToOnePartitioning]
+ endpoints(op.channels) shouldBe Seq(("w1", "u1"), ("w2", "u2"), ("w3", "u3"))
+ }
+
+ it should "silently truncate when from and to lengths differ (current behavior)" in {
+ // Pin: same `List.zip` truncation hazard as ChannelConfig (Bug #4799).
+ // Documenting the parallel here so a fix that aligns the two helpers
+ // surfaces this spec at the same time.
+ val out = LinkConfig.toPartitioning(
+ List(w1, w2, w3),
+ List(u1, u2),
+ OneToOnePartition(),
+ batch
+ )
+ val op = out.asInstanceOf[OneToOnePartitioning]
+ endpoints(op.channels) shouldBe Seq(("w1", "u1"), ("w2", "u2"))
+ }
+
+ // ----- BroadcastPartition -----
+
+ "BroadcastPartition" should "produce a BroadcastPartitioning whose channels follow zip pairing today (current behavior)" in {
+ // Pin: BroadcastPartition currently uses `fromWorkerIds.zip(toWorkerIds)`
+ // — the SAME 1:1 pairing as OneToOnePartition. ChannelConfig in the same
+ // package emits a full cross product for the BroadcastPartition arm,
+ // which matches broadcast semantics ("each sender targets every
+ // receiver"). The two helpers diverge today; pinning this so a fix that
+ // realigns the contract surfaces here. Filed as a Bug.
+ val out = LinkConfig.toPartitioning(
+ List(w1, w2, w3),
+ List(u1, u2, u3),
+ BroadcastPartition(),
+ batch
+ )
+ out shouldBe a[BroadcastPartitioning]
+ val bp = out.asInstanceOf[BroadcastPartitioning]
+ bp.batchSize shouldBe batch
+ endpoints(bp.channels) shouldBe Seq(("w1", "u1"), ("w2", "u2"), ("w3", "u3"))
+ }
+
+ it should "silently truncate broadcast pairings when sides differ in length (current behavior)" in {
+ val out = LinkConfig.toPartitioning(
+ List(w1, w2, w3),
+ List(u1, u2),
+ BroadcastPartition(),
+ batch
+ )
+ val bp = out.asInstanceOf[BroadcastPartitioning]
+ endpoints(bp.channels) shouldBe Seq(("w1", "u1"), ("w2", "u2"))
+ }
+
+ // ----- UnknownPartition -----
+
+ "UnknownPartition" should "produce a RoundRobinPartitioning with the full cross product" in {
+ val out = LinkConfig.toPartitioning(
+ List(w1, w2),
+ List(u1, u2),
+ UnknownPartition(),
+ batch
+ )
+ out shouldBe a[RoundRobinPartitioning]
+ val rr = out.asInstanceOf[RoundRobinPartitioning]
+ rr.batchSize shouldBe batch
+ endpoints(rr.channels) shouldBe Seq(
+ ("w1", "u1"),
+ ("w1", "u2"),
+ ("w2", "u1"),
+ ("w2", "u2")
+ )
+ }
+
+ // ----- empty inputs -----
+
+ it should "return empty channels when fromWorkerIds is empty (cross-product arm)" in {
+ val out = LinkConfig.toPartitioning(
+ Nil,
+ List(u1, u2),
+ HashPartition(),
+ batch
+ )
+ out.asInstanceOf[HashBasedShufflePartitioning].channels shouldBe empty
+ }
+
+ it should "return empty channels when toWorkerIds is empty (cross-product arm)" in {
+ val out = LinkConfig.toPartitioning(
+ List(w1, w2),
+ Nil,
+ HashPartition(),
+ batch
+ )
+ out.asInstanceOf[HashBasedShufflePartitioning].channels shouldBe empty
+ }
+
+ // ----- batch size propagation -----
+
+ it should "propagate dataTransferBatchSize verbatim regardless of partitioning arm" in {
+ val customBatch = 1024
+ val out = LinkConfig.toPartitioning(
Review Comment:
The three test cases starting at "empty inputs" use `it should ...` immediately after the "UnknownPartition" block, so their reported names become "UnknownPartition should ..." even though they are exercising HashPartition/OneToOnePartition behavior. Consider giving these tests an explicit subject string (e.g., "toPartitioning" should ...) or using `behavior of` to avoid misleading suite output.
##########
amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/config/LinkConfigSpec.scala:
##########
@@ -0,0 +1,231 @@
+/*
+ * 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.architecture.scheduling.config
+
+import org.apache.texera.amber.core.virtualidentity.{ActorVirtualIdentity, ChannelIdentity}
+import org.apache.texera.amber.core.workflow.{
+ BroadcastPartition,
+ HashPartition,
+ OneToOnePartition,
+ RangePartition,
+ SinglePartition,
+ UnknownPartition
+}
+import org.apache.texera.amber.engine.architecture.sendsemantics.partitionings.{
+ BroadcastPartitioning,
+ HashBasedShufflePartitioning,
+ OneToOnePartitioning,
+ RangeBasedShufflePartitioning,
+ RoundRobinPartitioning
+}
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+class LinkConfigSpec extends AnyFlatSpec with Matchers {
+
+ private val w1 = ActorVirtualIdentity("w1")
+ private val w2 = ActorVirtualIdentity("w2")
+ private val w3 = ActorVirtualIdentity("w3")
+ private val u1 = ActorVirtualIdentity("u1")
+ private val u2 = ActorVirtualIdentity("u2")
+ private val u3 = ActorVirtualIdentity("u3")
+ private val batch = 64
+
+ private def endpoints(channels: Seq[ChannelIdentity]): Seq[(String, String)] =
+ channels.map(c => (c.fromWorkerId.name, c.toWorkerId.name))
+
+ // ----- HashPartition -----
+
+ "toPartitioning" should "produce a HashBasedShufflePartitioning with full cross product channels" in {
+ val out = LinkConfig.toPartitioning(
+ List(w1, w2),
+ List(u1, u2, u3),
+ HashPartition(List("k1", "k2")),
+ batch
+ )
+ out shouldBe a[HashBasedShufflePartitioning]
+ val hp = out.asInstanceOf[HashBasedShufflePartitioning]
+ hp.batchSize shouldBe batch
+ hp.hashAttributeNames shouldBe Seq("k1", "k2")
+ endpoints(hp.channels) shouldBe Seq(
+ ("w1", "u1"),
+ ("w1", "u2"),
+ ("w1", "u3"),
+ ("w2", "u1"),
+ ("w2", "u2"),
+ ("w2", "u3")
+ )
+ hp.channels.foreach(_.isControl shouldBe false)
+ }
+
+ // ----- RangePartition -----
+
+ "RangePartition" should "produce a RangeBasedShufflePartitioning carrying the range bounds and cross-product channels" in {
+ val out = LinkConfig.toPartitioning(
+ List(w1),
+ List(u1, u2),
+ RangePartition(List("k"), 0L, 100L),
+ batch
+ )
+ out shouldBe a[RangeBasedShufflePartitioning]
+ val rp = out.asInstanceOf[RangeBasedShufflePartitioning]
+ rp.batchSize shouldBe batch
+ rp.rangeAttributeNames shouldBe Seq("k")
+ rp.rangeMin shouldBe 0L
+ rp.rangeMax shouldBe 100L
+ endpoints(rp.channels) shouldBe Seq(("w1", "u1"), ("w1", "u2"))
+ }
+
+ // ----- SinglePartition -----
+
+ "SinglePartition" should "produce a OneToOnePartitioning with one channel per from-worker to the single to-worker" in {
+ val out = LinkConfig.toPartitioning(
+ List(w1, w2, w3),
+ List(u1),
+ SinglePartition(),
+ batch
+ )
+ out shouldBe a[OneToOnePartitioning]
+ val op = out.asInstanceOf[OneToOnePartitioning]
+ op.batchSize shouldBe batch
+ endpoints(op.channels) shouldBe Seq(("w1", "u1"), ("w2", "u1"), ("w3", "u1"))
+ }
+
+ it should "raise an AssertionError when more than one to-worker is supplied" in {
+ assertThrows[AssertionError] {
+ LinkConfig.toPartitioning(List(w1, w2), List(u1, u2), SinglePartition(), batch)
+ }
+ }
+
+ // ----- OneToOnePartition -----
+
+ "OneToOnePartition" should "produce a OneToOnePartitioning with zip pairing for equal-length inputs" in {
+ val out = LinkConfig.toPartitioning(
+ List(w1, w2, w3),
+ List(u1, u2, u3),
+ OneToOnePartition(),
+ batch
+ )
+ out shouldBe a[OneToOnePartitioning]
+ val op = out.asInstanceOf[OneToOnePartitioning]
+ endpoints(op.channels) shouldBe Seq(("w1", "u1"), ("w2", "u2"), ("w3", "u3"))
+ }
+
+ it should "silently truncate when from and to lengths differ (current behavior)" in {
+ // Pin: same `List.zip` truncation hazard as ChannelConfig (Bug #4799).
+ // Documenting the parallel here so a fix that aligns the two helpers
+ // surfaces this spec at the same time.
+ val out = LinkConfig.toPartitioning(
+ List(w1, w2, w3),
+ List(u1, u2),
+ OneToOnePartition(),
+ batch
+ )
+ val op = out.asInstanceOf[OneToOnePartitioning]
+ endpoints(op.channels) shouldBe Seq(("w1", "u1"), ("w2", "u2"))
+ }
+
+ // ----- BroadcastPartition -----
+
+ "BroadcastPartition" should "produce a BroadcastPartitioning whose channels follow zip pairing today (current behavior)" in {
+ // Pin: BroadcastPartition currently uses `fromWorkerIds.zip(toWorkerIds)`
+ // — the SAME 1:1 pairing as OneToOnePartition. ChannelConfig in the same
+ // package emits a full cross product for the BroadcastPartition arm,
+ // which matches broadcast semantics ("each sender targets every
+ // receiver"). The two helpers diverge today; pinning this so a fix that
+ // realigns the contract surfaces here. Filed as a Bug.
+ val out = LinkConfig.toPartitioning(
+ List(w1, w2, w3),
+ List(u1, u2, u3),
+ BroadcastPartition(),
+ batch
+ )
+ out shouldBe a[BroadcastPartitioning]
+ val bp = out.asInstanceOf[BroadcastPartitioning]
+ bp.batchSize shouldBe batch
+ endpoints(bp.channels) shouldBe Seq(("w1", "u1"), ("w2", "u2"), ("w3", "u3"))
+ }
+
+ it should "silently truncate broadcast pairings when sides differ in length (current behavior)" in {
+ val out = LinkConfig.toPartitioning(
+ List(w1, w2, w3),
+ List(u1, u2),
+ BroadcastPartition(),
+ batch
+ )
+ val bp = out.asInstanceOf[BroadcastPartitioning]
+ endpoints(bp.channels) shouldBe Seq(("w1", "u1"), ("w2", "u2"))
+ }
+
+ // ----- UnknownPartition -----
+
+ "UnknownPartition" should "produce a RoundRobinPartitioning with the full cross product" in {
+ val out = LinkConfig.toPartitioning(
+ List(w1, w2),
+ List(u1, u2),
+ UnknownPartition(),
+ batch
+ )
+ out shouldBe a[RoundRobinPartitioning]
+ val rr = out.asInstanceOf[RoundRobinPartitioning]
+ rr.batchSize shouldBe batch
+ endpoints(rr.channels) shouldBe Seq(
+ ("w1", "u1"),
+ ("w1", "u2"),
+ ("w2", "u1"),
+ ("w2", "u2")
+ )
+ }
+
Review Comment:
`LinkConfig.toPartitioning` has a `case _ => throw new UnsupportedOperationException()` branch, but the spec doesn't currently assert that behavior. Since `PartitionInfo` is sealed (so you can't create a new subtype in tests), the simplest way to cover this branch is to pass `null` as the `partitionInfo` and assert `UnsupportedOperationException` is thrown.
--
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 unit tests for LinkConfig.toPartitioning [texera]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #4803:
URL: https://github.com/apache/texera/pull/4803#issuecomment-4365252304
## [Codecov](https://app.codecov.io/gh/apache/texera/pull/4803?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 42.73%. Comparing base ([`88c7a21`](https://app.codecov.io/gh/apache/texera/commit/88c7a21ea15c6e440de54786e31700b3a9cf7c22?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`a2acc2e`](https://app.codecov.io/gh/apache/texera/commit/a2acc2ea47bf28a1604b3b088883d67bd68a1409?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## main #4803 +/- ##
=========================================
Coverage 42.73% 42.73%
Complexity 2003 2003
=========================================
Files 957 957
Lines 34077 34077
Branches 3753 3753
=========================================
Hits 14564 14564
Misses 18733 18733
Partials 780 780
```
| [Flag](https://app.codecov.io/gh/apache/texera/pull/4803/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [python](https://app.codecov.io/gh/apache/texera/pull/4803/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `84.15% <ø> (ø)` | |
Flags with carried forward coverage won't be shown. [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#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/texera/pull/4803?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>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :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 unit tests for LinkConfig.toPartitioning [texera]
Posted by "Yicong-Huang (via GitHub)" <gi...@apache.org>.
Yicong-Huang commented on code in PR #4803:
URL: https://github.com/apache/texera/pull/4803#discussion_r3177550744
##########
amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/config/LinkConfigSpec.scala:
##########
@@ -0,0 +1,231 @@
+/*
+ * 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.architecture.scheduling.config
+
+import org.apache.texera.amber.core.virtualidentity.{ActorVirtualIdentity, ChannelIdentity}
+import org.apache.texera.amber.core.workflow.{
+ BroadcastPartition,
+ HashPartition,
+ OneToOnePartition,
+ RangePartition,
+ SinglePartition,
+ UnknownPartition
+}
+import org.apache.texera.amber.engine.architecture.sendsemantics.partitionings.{
+ BroadcastPartitioning,
+ HashBasedShufflePartitioning,
+ OneToOnePartitioning,
+ RangeBasedShufflePartitioning,
+ RoundRobinPartitioning
+}
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+class LinkConfigSpec extends AnyFlatSpec with Matchers {
+
+ private val w1 = ActorVirtualIdentity("w1")
+ private val w2 = ActorVirtualIdentity("w2")
+ private val w3 = ActorVirtualIdentity("w3")
+ private val u1 = ActorVirtualIdentity("u1")
+ private val u2 = ActorVirtualIdentity("u2")
+ private val u3 = ActorVirtualIdentity("u3")
+ private val batch = 64
+
+ private def endpoints(channels: Seq[ChannelIdentity]): Seq[(String, String)] =
+ channels.map(c => (c.fromWorkerId.name, c.toWorkerId.name))
+
+ // ----- HashPartition -----
+
+ "toPartitioning" should "produce a HashBasedShufflePartitioning with full cross product channels" in {
+ val out = LinkConfig.toPartitioning(
+ List(w1, w2),
+ List(u1, u2, u3),
+ HashPartition(List("k1", "k2")),
+ batch
+ )
+ out shouldBe a[HashBasedShufflePartitioning]
+ val hp = out.asInstanceOf[HashBasedShufflePartitioning]
+ hp.batchSize shouldBe batch
+ hp.hashAttributeNames shouldBe Seq("k1", "k2")
+ endpoints(hp.channels) shouldBe Seq(
+ ("w1", "u1"),
+ ("w1", "u2"),
+ ("w1", "u3"),
+ ("w2", "u1"),
+ ("w2", "u2"),
+ ("w2", "u3")
+ )
+ hp.channels.foreach(_.isControl shouldBe false)
+ }
+
+ // ----- RangePartition -----
+
+ "RangePartition" should "produce a RangeBasedShufflePartitioning carrying the range bounds and cross-product channels" in {
+ val out = LinkConfig.toPartitioning(
+ List(w1),
+ List(u1, u2),
+ RangePartition(List("k"), 0L, 100L),
+ batch
+ )
+ out shouldBe a[RangeBasedShufflePartitioning]
+ val rp = out.asInstanceOf[RangeBasedShufflePartitioning]
+ rp.batchSize shouldBe batch
+ rp.rangeAttributeNames shouldBe Seq("k")
+ rp.rangeMin shouldBe 0L
+ rp.rangeMax shouldBe 100L
+ endpoints(rp.channels) shouldBe Seq(("w1", "u1"), ("w1", "u2"))
+ }
+
+ // ----- SinglePartition -----
+
+ "SinglePartition" should "produce a OneToOnePartitioning with one channel per from-worker to the single to-worker" in {
+ val out = LinkConfig.toPartitioning(
+ List(w1, w2, w3),
+ List(u1),
+ SinglePartition(),
+ batch
+ )
+ out shouldBe a[OneToOnePartitioning]
+ val op = out.asInstanceOf[OneToOnePartitioning]
+ op.batchSize shouldBe batch
+ endpoints(op.channels) shouldBe Seq(("w1", "u1"), ("w2", "u1"), ("w3", "u1"))
+ }
+
+ it should "raise an AssertionError when more than one to-worker is supplied" in {
+ assertThrows[AssertionError] {
+ LinkConfig.toPartitioning(List(w1, w2), List(u1, u2), SinglePartition(), batch)
+ }
+ }
+
+ // ----- OneToOnePartition -----
+
+ "OneToOnePartition" should "produce a OneToOnePartitioning with zip pairing for equal-length inputs" in {
+ val out = LinkConfig.toPartitioning(
+ List(w1, w2, w3),
+ List(u1, u2, u3),
+ OneToOnePartition(),
+ batch
+ )
+ out shouldBe a[OneToOnePartitioning]
+ val op = out.asInstanceOf[OneToOnePartitioning]
+ endpoints(op.channels) shouldBe Seq(("w1", "u1"), ("w2", "u2"), ("w3", "u3"))
+ }
+
+ it should "silently truncate when from and to lengths differ (current behavior)" in {
+ // Pin: same `List.zip` truncation hazard as ChannelConfig (Bug #4799).
+ // Documenting the parallel here so a fix that aligns the two helpers
+ // surfaces this spec at the same time.
+ val out = LinkConfig.toPartitioning(
+ List(w1, w2, w3),
+ List(u1, u2),
+ OneToOnePartition(),
+ batch
+ )
+ val op = out.asInstanceOf[OneToOnePartitioning]
+ endpoints(op.channels) shouldBe Seq(("w1", "u1"), ("w2", "u2"))
+ }
+
+ // ----- BroadcastPartition -----
+
+ "BroadcastPartition" should "produce a BroadcastPartitioning whose channels follow zip pairing today (current behavior)" in {
+ // Pin: BroadcastPartition currently uses `fromWorkerIds.zip(toWorkerIds)`
+ // — the SAME 1:1 pairing as OneToOnePartition. ChannelConfig in the same
+ // package emits a full cross product for the BroadcastPartition arm,
+ // which matches broadcast semantics ("each sender targets every
+ // receiver"). The two helpers diverge today; pinning this so a fix that
+ // realigns the contract surfaces here. Filed as a Bug.
+ val out = LinkConfig.toPartitioning(
+ List(w1, w2, w3),
+ List(u1, u2, u3),
+ BroadcastPartition(),
+ batch
+ )
+ out shouldBe a[BroadcastPartitioning]
+ val bp = out.asInstanceOf[BroadcastPartitioning]
+ bp.batchSize shouldBe batch
+ endpoints(bp.channels) shouldBe Seq(("w1", "u1"), ("w2", "u2"), ("w3", "u3"))
+ }
+
+ it should "silently truncate broadcast pairings when sides differ in length (current behavior)" in {
+ val out = LinkConfig.toPartitioning(
+ List(w1, w2, w3),
+ List(u1, u2),
+ BroadcastPartition(),
+ batch
+ )
+ val bp = out.asInstanceOf[BroadcastPartitioning]
+ endpoints(bp.channels) shouldBe Seq(("w1", "u1"), ("w2", "u2"))
+ }
+
+ // ----- UnknownPartition -----
+
+ "UnknownPartition" should "produce a RoundRobinPartitioning with the full cross product" in {
+ val out = LinkConfig.toPartitioning(
+ List(w1, w2),
+ List(u1, u2),
+ UnknownPartition(),
+ batch
+ )
+ out shouldBe a[RoundRobinPartitioning]
+ val rr = out.asInstanceOf[RoundRobinPartitioning]
+ rr.batchSize shouldBe batch
+ endpoints(rr.channels) shouldBe Seq(
+ ("w1", "u1"),
+ ("w1", "u2"),
+ ("w2", "u1"),
+ ("w2", "u2")
+ )
+ }
+
+ // ----- empty inputs -----
+
+ it should "return empty channels when fromWorkerIds is empty (cross-product arm)" in {
+ val out = LinkConfig.toPartitioning(
+ Nil,
+ List(u1, u2),
+ HashPartition(),
+ batch
+ )
+ out.asInstanceOf[HashBasedShufflePartitioning].channels shouldBe empty
+ }
+
+ it should "return empty channels when toWorkerIds is empty (cross-product arm)" in {
+ val out = LinkConfig.toPartitioning(
+ List(w1, w2),
+ Nil,
+ HashPartition(),
+ batch
+ )
+ out.asInstanceOf[HashBasedShufflePartitioning].channels shouldBe empty
+ }
+
+ // ----- batch size propagation -----
+
+ it should "propagate dataTransferBatchSize verbatim regardless of partitioning arm" in {
+ val customBatch = 1024
+ val out = LinkConfig.toPartitioning(
Review Comment:
Re-anchored the trailing tests in b64e24d99e: the empty-input and batch-propagation cases now start a new `"toPartitioning" should ...` subject (and the new unsupported-branch case also lives there), so ScalaTest no longer reports them as `"UnknownPartition" should ...`.
##########
amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/config/LinkConfigSpec.scala:
##########
@@ -0,0 +1,231 @@
+/*
+ * 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.architecture.scheduling.config
+
+import org.apache.texera.amber.core.virtualidentity.{ActorVirtualIdentity, ChannelIdentity}
+import org.apache.texera.amber.core.workflow.{
+ BroadcastPartition,
+ HashPartition,
+ OneToOnePartition,
+ RangePartition,
+ SinglePartition,
+ UnknownPartition
+}
+import org.apache.texera.amber.engine.architecture.sendsemantics.partitionings.{
+ BroadcastPartitioning,
+ HashBasedShufflePartitioning,
+ OneToOnePartitioning,
+ RangeBasedShufflePartitioning,
+ RoundRobinPartitioning
+}
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+class LinkConfigSpec extends AnyFlatSpec with Matchers {
+
+ private val w1 = ActorVirtualIdentity("w1")
+ private val w2 = ActorVirtualIdentity("w2")
+ private val w3 = ActorVirtualIdentity("w3")
+ private val u1 = ActorVirtualIdentity("u1")
+ private val u2 = ActorVirtualIdentity("u2")
+ private val u3 = ActorVirtualIdentity("u3")
+ private val batch = 64
+
+ private def endpoints(channels: Seq[ChannelIdentity]): Seq[(String, String)] =
+ channels.map(c => (c.fromWorkerId.name, c.toWorkerId.name))
+
+ // ----- HashPartition -----
+
+ "toPartitioning" should "produce a HashBasedShufflePartitioning with full cross product channels" in {
+ val out = LinkConfig.toPartitioning(
+ List(w1, w2),
+ List(u1, u2, u3),
+ HashPartition(List("k1", "k2")),
+ batch
+ )
+ out shouldBe a[HashBasedShufflePartitioning]
+ val hp = out.asInstanceOf[HashBasedShufflePartitioning]
+ hp.batchSize shouldBe batch
+ hp.hashAttributeNames shouldBe Seq("k1", "k2")
+ endpoints(hp.channels) shouldBe Seq(
+ ("w1", "u1"),
+ ("w1", "u2"),
+ ("w1", "u3"),
+ ("w2", "u1"),
+ ("w2", "u2"),
+ ("w2", "u3")
+ )
+ hp.channels.foreach(_.isControl shouldBe false)
+ }
+
+ // ----- RangePartition -----
+
+ "RangePartition" should "produce a RangeBasedShufflePartitioning carrying the range bounds and cross-product channels" in {
+ val out = LinkConfig.toPartitioning(
+ List(w1),
+ List(u1, u2),
+ RangePartition(List("k"), 0L, 100L),
+ batch
+ )
+ out shouldBe a[RangeBasedShufflePartitioning]
+ val rp = out.asInstanceOf[RangeBasedShufflePartitioning]
+ rp.batchSize shouldBe batch
+ rp.rangeAttributeNames shouldBe Seq("k")
+ rp.rangeMin shouldBe 0L
+ rp.rangeMax shouldBe 100L
+ endpoints(rp.channels) shouldBe Seq(("w1", "u1"), ("w1", "u2"))
+ }
+
+ // ----- SinglePartition -----
+
+ "SinglePartition" should "produce a OneToOnePartitioning with one channel per from-worker to the single to-worker" in {
+ val out = LinkConfig.toPartitioning(
+ List(w1, w2, w3),
+ List(u1),
+ SinglePartition(),
+ batch
+ )
+ out shouldBe a[OneToOnePartitioning]
+ val op = out.asInstanceOf[OneToOnePartitioning]
+ op.batchSize shouldBe batch
+ endpoints(op.channels) shouldBe Seq(("w1", "u1"), ("w2", "u1"), ("w3", "u1"))
+ }
+
+ it should "raise an AssertionError when more than one to-worker is supplied" in {
+ assertThrows[AssertionError] {
+ LinkConfig.toPartitioning(List(w1, w2), List(u1, u2), SinglePartition(), batch)
+ }
+ }
+
+ // ----- OneToOnePartition -----
+
+ "OneToOnePartition" should "produce a OneToOnePartitioning with zip pairing for equal-length inputs" in {
+ val out = LinkConfig.toPartitioning(
+ List(w1, w2, w3),
+ List(u1, u2, u3),
+ OneToOnePartition(),
+ batch
+ )
+ out shouldBe a[OneToOnePartitioning]
+ val op = out.asInstanceOf[OneToOnePartitioning]
+ endpoints(op.channels) shouldBe Seq(("w1", "u1"), ("w2", "u2"), ("w3", "u3"))
+ }
+
+ it should "silently truncate when from and to lengths differ (current behavior)" in {
+ // Pin: same `List.zip` truncation hazard as ChannelConfig (Bug #4799).
+ // Documenting the parallel here so a fix that aligns the two helpers
+ // surfaces this spec at the same time.
+ val out = LinkConfig.toPartitioning(
+ List(w1, w2, w3),
+ List(u1, u2),
+ OneToOnePartition(),
+ batch
+ )
+ val op = out.asInstanceOf[OneToOnePartitioning]
+ endpoints(op.channels) shouldBe Seq(("w1", "u1"), ("w2", "u2"))
+ }
+
+ // ----- BroadcastPartition -----
+
+ "BroadcastPartition" should "produce a BroadcastPartitioning whose channels follow zip pairing today (current behavior)" in {
+ // Pin: BroadcastPartition currently uses `fromWorkerIds.zip(toWorkerIds)`
+ // — the SAME 1:1 pairing as OneToOnePartition. ChannelConfig in the same
+ // package emits a full cross product for the BroadcastPartition arm,
+ // which matches broadcast semantics ("each sender targets every
+ // receiver"). The two helpers diverge today; pinning this so a fix that
+ // realigns the contract surfaces here. Filed as a Bug.
+ val out = LinkConfig.toPartitioning(
+ List(w1, w2, w3),
+ List(u1, u2, u3),
+ BroadcastPartition(),
+ batch
+ )
+ out shouldBe a[BroadcastPartitioning]
+ val bp = out.asInstanceOf[BroadcastPartitioning]
+ bp.batchSize shouldBe batch
+ endpoints(bp.channels) shouldBe Seq(("w1", "u1"), ("w2", "u2"), ("w3", "u3"))
+ }
+
+ it should "silently truncate broadcast pairings when sides differ in length (current behavior)" in {
+ val out = LinkConfig.toPartitioning(
+ List(w1, w2, w3),
+ List(u1, u2),
+ BroadcastPartition(),
+ batch
+ )
+ val bp = out.asInstanceOf[BroadcastPartitioning]
+ endpoints(bp.channels) shouldBe Seq(("w1", "u1"), ("w2", "u2"))
+ }
+
+ // ----- UnknownPartition -----
+
+ "UnknownPartition" should "produce a RoundRobinPartitioning with the full cross product" in {
+ val out = LinkConfig.toPartitioning(
+ List(w1, w2),
+ List(u1, u2),
+ UnknownPartition(),
+ batch
+ )
+ out shouldBe a[RoundRobinPartitioning]
+ val rr = out.asInstanceOf[RoundRobinPartitioning]
+ rr.batchSize shouldBe batch
+ endpoints(rr.channels) shouldBe Seq(
+ ("w1", "u1"),
+ ("w1", "u2"),
+ ("w2", "u1"),
+ ("w2", "u2")
+ )
+ }
+
Review Comment:
Added in b64e24d99e: a new `it should "throw UnsupportedOperationException when partitionInfo is unrecognized"` test passes `null` as the `partitionInfo` (since `PartitionInfo` is sealed, this is the only off-domain value reachable from a test) and asserts `UnsupportedOperationException`. The catch-all branch is now pinned.
--
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