aglinxinyuan commented on code in PR #4746: URL: https://github.com/apache/texera/pull/4746#discussion_r3177512407
########## amber/src/test/scala/org/apache/texera/amber/engine/architecture/sendsemantics/partitioners/PartitionersSpec.scala: ########## @@ -0,0 +1,190 @@ +/* + * 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.sendsemantics.partitioners + +import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple} +import org.apache.texera.amber.core.virtualidentity.{ActorVirtualIdentity, ChannelIdentity} +import org.apache.texera.amber.engine.architecture.sendsemantics.partitionings.{ + BroadcastPartitioning, + HashBasedShufflePartitioning, + OneToOnePartitioning, + RoundRobinPartitioning +} +import org.scalatest.flatspec.AnyFlatSpec + +class PartitionersSpec extends AnyFlatSpec { + + private val sender: ActorVirtualIdentity = ActorVirtualIdentity("sender") + private val r1: ActorVirtualIdentity = ActorVirtualIdentity("rec1") + private val r2: ActorVirtualIdentity = ActorVirtualIdentity("rec2") + private val r3: ActorVirtualIdentity = ActorVirtualIdentity("rec3") + + private def channel(to: ActorVirtualIdentity): ChannelIdentity = + ChannelIdentity(sender, to, isControl = false) + + private val intAttr: Attribute = new Attribute("v", AttributeType.INTEGER) + private val intSchema: Schema = Schema().add(intAttr) + + private def intTuple(value: Int): Tuple = + Tuple.builder(intSchema).add(intAttr, value).build() + + private val twoStringSchema: Schema = Schema() + .add(new Attribute("k", AttributeType.STRING)) + .add(new Attribute("v", AttributeType.STRING)) + + private def stringTuple(k: String, v: String): Tuple = + Tuple + .builder(twoStringSchema) + .add(new Attribute("k", AttributeType.STRING), k) + .add(new Attribute("v", AttributeType.STRING), v) + .build() + + // -- OneToOnePartitioner -------------------------------------------------- + + "OneToOnePartitioner.getBucketIndex" should "always return Iterator(0)" in { + val partitioning = OneToOnePartitioning( + batchSize = 100, + channels = Seq(channel(r1)) + ) + val partitioner = OneToOnePartitioner(partitioning, sender) + assert(partitioner.getBucketIndex(intTuple(7)).toList == List(0)) + assert(partitioner.getBucketIndex(intTuple(42)).toList == List(0)) + } + + "OneToOnePartitioner.allReceivers" should "return the receiver from the channel matching the actor id" in { + val partitioning = OneToOnePartitioning( + batchSize = 100, + channels = Seq( + ChannelIdentity(ActorVirtualIdentity("other-sender"), r2, isControl = false), + channel(r1) + ) + ) + val partitioner = OneToOnePartitioner(partitioning, sender) + assert(partitioner.allReceivers == Seq(r1)) + } + + // -- BroadcastPartitioner ------------------------------------------------- + + "BroadcastPartitioner.getBucketIndex" should "yield every receiver index for any tuple" in { + val partitioning = BroadcastPartitioning( + batchSize = 100, + channels = Seq(channel(r1), channel(r2), channel(r3)) + ) + val partitioner = BroadcastPartitioner(partitioning) + assert(partitioner.getBucketIndex(intTuple(0)).toList == List(0, 1, 2)) + } + + "BroadcastPartitioner" should "deduplicate receivers when channels list a worker twice" in { + val partitioning = BroadcastPartitioning( + batchSize = 100, + channels = Seq(channel(r1), channel(r1), channel(r2)) + ) + val partitioner = BroadcastPartitioner(partitioning) + assert(partitioner.allReceivers == Seq(r1, r2)) + assert(partitioner.getBucketIndex(intTuple(0)).toList == List(0, 1)) + } + + // -- RoundRobinPartitioner ------------------------------------------------ + + "RoundRobinPartitioner.getBucketIndex" should "cycle through bucket indices" in { + val partitioning = RoundRobinPartitioning( + batchSize = 100, + channels = Seq(channel(r1), channel(r2), channel(r3)) + ) + val partitioner = RoundRobinPartitioner(partitioning) + + val indices = (1 to 7).map(_ => partitioner.getBucketIndex(intTuple(0)).next()).toList + // Implementation increments first, then emits. Starting from 0, the first + // emitted index is therefore 1, then 2, then 0, repeating. + assert(indices == List(1, 2, 0, 1, 2, 0, 1)) + } + + "RoundRobinPartitioner.allReceivers" should "preserve channel order while deduplicating" in { + val partitioning = RoundRobinPartitioning( + batchSize = 100, + channels = Seq(channel(r2), channel(r1), channel(r2)) + ) + val partitioner = RoundRobinPartitioner(partitioning) + assert(partitioner.allReceivers == Seq(r2, r1)) + } + + // -- HashBasedShufflePartitioner ------------------------------------------ + + "HashBasedShufflePartitioner.getBucketIndex" should "return a non-negative index within the receiver count" in { + val partitioning = HashBasedShufflePartitioning( + batchSize = 100, + channels = Seq(channel(r1), channel(r2), channel(r3)), + hashAttributeNames = Seq("k") + ) + val partitioner = HashBasedShufflePartitioner(partitioning) + + (0 until 50).foreach { i => + val idx = partitioner.getBucketIndex(stringTuple(s"key-$i", "v")).next() + assert(idx >= 0 && idx < 3, s"index $idx out of range for tuple key-$i") + } + } + + it should "be deterministic for the same hash-key value" in { + val partitioning = HashBasedShufflePartitioning( + batchSize = 100, + channels = Seq(channel(r1), channel(r2), channel(r3)), + hashAttributeNames = Seq("k") + ) + val partitioner = HashBasedShufflePartitioner(partitioning) + + val first = partitioner.getBucketIndex(stringTuple("alpha", "ignored-1")).next() + val second = partitioner.getBucketIndex(stringTuple("alpha", "ignored-2")).next() + assert(first == second) + } + + it should "depend only on the hash-attribute subset, not on other fields" in { + val partitioning = HashBasedShufflePartitioning( + batchSize = 100, + channels = Seq(channel(r1), channel(r2), channel(r3)), + hashAttributeNames = Seq("k") + ) + val partitioner = HashBasedShufflePartitioner(partitioning) + + val a = partitioner.getBucketIndex(stringTuple("same-key", "value-A")).next() + val b = partitioner.getBucketIndex(stringTuple("same-key", "value-B")).next() + assert(a == b) + } + + it should "use the full tuple when no hash attributes are configured" in { + val partitioning = HashBasedShufflePartitioning( + batchSize = 100, + channels = Seq(channel(r1), channel(r2), channel(r3)), + hashAttributeNames = Seq.empty + ) + val partitioner = HashBasedShufflePartitioner(partitioning) + val idx = partitioner.getBucketIndex(stringTuple("k", "v")).next() + assert(idx >= 0 && idx < 3) Review Comment: Done in b3d118031e. The test now drives 50 tuples that hold `k` constant but vary `v` and asserts the resulting buckets span more than one value. An implementation that ignored `v` (or returned a constant) would fail. ########## amber/src/test/scala/org/apache/texera/amber/engine/architecture/sendsemantics/partitioners/PartitionersSpec.scala: ########## @@ -0,0 +1,190 @@ +/* + * 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.sendsemantics.partitioners + +import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple} +import org.apache.texera.amber.core.virtualidentity.{ActorVirtualIdentity, ChannelIdentity} +import org.apache.texera.amber.engine.architecture.sendsemantics.partitionings.{ + BroadcastPartitioning, + HashBasedShufflePartitioning, + OneToOnePartitioning, + RoundRobinPartitioning +} +import org.scalatest.flatspec.AnyFlatSpec + +class PartitionersSpec extends AnyFlatSpec { + + private val sender: ActorVirtualIdentity = ActorVirtualIdentity("sender") + private val r1: ActorVirtualIdentity = ActorVirtualIdentity("rec1") + private val r2: ActorVirtualIdentity = ActorVirtualIdentity("rec2") + private val r3: ActorVirtualIdentity = ActorVirtualIdentity("rec3") + + private def channel(to: ActorVirtualIdentity): ChannelIdentity = + ChannelIdentity(sender, to, isControl = false) + + private val intAttr: Attribute = new Attribute("v", AttributeType.INTEGER) + private val intSchema: Schema = Schema().add(intAttr) + + private def intTuple(value: Int): Tuple = + Tuple.builder(intSchema).add(intAttr, value).build() + + private val twoStringSchema: Schema = Schema() + .add(new Attribute("k", AttributeType.STRING)) + .add(new Attribute("v", AttributeType.STRING)) + + private def stringTuple(k: String, v: String): Tuple = + Tuple + .builder(twoStringSchema) + .add(new Attribute("k", AttributeType.STRING), k) + .add(new Attribute("v", AttributeType.STRING), v) + .build() + + // -- OneToOnePartitioner -------------------------------------------------- + + "OneToOnePartitioner.getBucketIndex" should "always return Iterator(0)" in { + val partitioning = OneToOnePartitioning( + batchSize = 100, + channels = Seq(channel(r1)) + ) + val partitioner = OneToOnePartitioner(partitioning, sender) + assert(partitioner.getBucketIndex(intTuple(7)).toList == List(0)) + assert(partitioner.getBucketIndex(intTuple(42)).toList == List(0)) + } + + "OneToOnePartitioner.allReceivers" should "return the receiver from the channel matching the actor id" in { + val partitioning = OneToOnePartitioning( + batchSize = 100, + channels = Seq( + ChannelIdentity(ActorVirtualIdentity("other-sender"), r2, isControl = false), + channel(r1) + ) + ) + val partitioner = OneToOnePartitioner(partitioning, sender) + assert(partitioner.allReceivers == Seq(r1)) + } + + // -- BroadcastPartitioner ------------------------------------------------- + + "BroadcastPartitioner.getBucketIndex" should "yield every receiver index for any tuple" in { + val partitioning = BroadcastPartitioning( + batchSize = 100, + channels = Seq(channel(r1), channel(r2), channel(r3)) + ) + val partitioner = BroadcastPartitioner(partitioning) + assert(partitioner.getBucketIndex(intTuple(0)).toList == List(0, 1, 2)) + } + + "BroadcastPartitioner" should "deduplicate receivers when channels list a worker twice" in { + val partitioning = BroadcastPartitioning( + batchSize = 100, + channels = Seq(channel(r1), channel(r1), channel(r2)) + ) + val partitioner = BroadcastPartitioner(partitioning) + assert(partitioner.allReceivers == Seq(r1, r2)) + assert(partitioner.getBucketIndex(intTuple(0)).toList == List(0, 1)) + } + + // -- RoundRobinPartitioner ------------------------------------------------ + + "RoundRobinPartitioner.getBucketIndex" should "cycle through bucket indices" in { + val partitioning = RoundRobinPartitioning( + batchSize = 100, + channels = Seq(channel(r1), channel(r2), channel(r3)) + ) + val partitioner = RoundRobinPartitioner(partitioning) + + val indices = (1 to 7).map(_ => partitioner.getBucketIndex(intTuple(0)).next()).toList + // Implementation increments first, then emits. Starting from 0, the first + // emitted index is therefore 1, then 2, then 0, repeating. + assert(indices == List(1, 2, 0, 1, 2, 0, 1)) + } + + "RoundRobinPartitioner.allReceivers" should "preserve channel order while deduplicating" in { + val partitioning = RoundRobinPartitioning( + batchSize = 100, + channels = Seq(channel(r2), channel(r1), channel(r2)) + ) + val partitioner = RoundRobinPartitioner(partitioning) + assert(partitioner.allReceivers == Seq(r2, r1)) + } + + // -- HashBasedShufflePartitioner ------------------------------------------ + + "HashBasedShufflePartitioner.getBucketIndex" should "return a non-negative index within the receiver count" in { + val partitioning = HashBasedShufflePartitioning( + batchSize = 100, + channels = Seq(channel(r1), channel(r2), channel(r3)), + hashAttributeNames = Seq("k") + ) + val partitioner = HashBasedShufflePartitioner(partitioning) + + (0 until 50).foreach { i => + val idx = partitioner.getBucketIndex(stringTuple(s"key-$i", "v")).next() + assert(idx >= 0 && idx < 3, s"index $idx out of range for tuple key-$i") + } + } + + it should "be deterministic for the same hash-key value" in { + val partitioning = HashBasedShufflePartitioning( + batchSize = 100, + channels = Seq(channel(r1), channel(r2), channel(r3)), + hashAttributeNames = Seq("k") + ) + val partitioner = HashBasedShufflePartitioner(partitioning) + + val first = partitioner.getBucketIndex(stringTuple("alpha", "ignored-1")).next() + val second = partitioner.getBucketIndex(stringTuple("alpha", "ignored-2")).next() Review Comment: Done in b3d118031e. Determinism is now exercised with the *same tuple instance* on two consecutive calls, so a partitioner with per-call state can no longer pass. ########## amber/src/test/scala/org/apache/texera/amber/engine/architecture/sendsemantics/partitioners/PartitionersSpec.scala: ########## @@ -0,0 +1,190 @@ +/* + * 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.sendsemantics.partitioners + +import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple} +import org.apache.texera.amber.core.virtualidentity.{ActorVirtualIdentity, ChannelIdentity} +import org.apache.texera.amber.engine.architecture.sendsemantics.partitionings.{ + BroadcastPartitioning, + HashBasedShufflePartitioning, + OneToOnePartitioning, + RoundRobinPartitioning +} +import org.scalatest.flatspec.AnyFlatSpec + +class PartitionersSpec extends AnyFlatSpec { + + private val sender: ActorVirtualIdentity = ActorVirtualIdentity("sender") + private val r1: ActorVirtualIdentity = ActorVirtualIdentity("rec1") + private val r2: ActorVirtualIdentity = ActorVirtualIdentity("rec2") + private val r3: ActorVirtualIdentity = ActorVirtualIdentity("rec3") + + private def channel(to: ActorVirtualIdentity): ChannelIdentity = + ChannelIdentity(sender, to, isControl = false) + + private val intAttr: Attribute = new Attribute("v", AttributeType.INTEGER) + private val intSchema: Schema = Schema().add(intAttr) + + private def intTuple(value: Int): Tuple = + Tuple.builder(intSchema).add(intAttr, value).build() + + private val twoStringSchema: Schema = Schema() + .add(new Attribute("k", AttributeType.STRING)) + .add(new Attribute("v", AttributeType.STRING)) + + private def stringTuple(k: String, v: String): Tuple = + Tuple + .builder(twoStringSchema) + .add(new Attribute("k", AttributeType.STRING), k) + .add(new Attribute("v", AttributeType.STRING), v) + .build() + + // -- OneToOnePartitioner -------------------------------------------------- + + "OneToOnePartitioner.getBucketIndex" should "always return Iterator(0)" in { + val partitioning = OneToOnePartitioning( + batchSize = 100, + channels = Seq(channel(r1)) + ) + val partitioner = OneToOnePartitioner(partitioning, sender) + assert(partitioner.getBucketIndex(intTuple(7)).toList == List(0)) + assert(partitioner.getBucketIndex(intTuple(42)).toList == List(0)) + } + + "OneToOnePartitioner.allReceivers" should "return the receiver from the channel matching the actor id" in { + val partitioning = OneToOnePartitioning( + batchSize = 100, + channels = Seq( + ChannelIdentity(ActorVirtualIdentity("other-sender"), r2, isControl = false), + channel(r1) + ) + ) + val partitioner = OneToOnePartitioner(partitioning, sender) + assert(partitioner.allReceivers == Seq(r1)) + } + + // -- BroadcastPartitioner ------------------------------------------------- + + "BroadcastPartitioner.getBucketIndex" should "yield every receiver index for any tuple" in { + val partitioning = BroadcastPartitioning( + batchSize = 100, + channels = Seq(channel(r1), channel(r2), channel(r3)) + ) + val partitioner = BroadcastPartitioner(partitioning) + assert(partitioner.getBucketIndex(intTuple(0)).toList == List(0, 1, 2)) + } + + "BroadcastPartitioner" should "deduplicate receivers when channels list a worker twice" in { + val partitioning = BroadcastPartitioning( + batchSize = 100, + channels = Seq(channel(r1), channel(r1), channel(r2)) + ) + val partitioner = BroadcastPartitioner(partitioning) + assert(partitioner.allReceivers == Seq(r1, r2)) + assert(partitioner.getBucketIndex(intTuple(0)).toList == List(0, 1)) + } + + // -- RoundRobinPartitioner ------------------------------------------------ + + "RoundRobinPartitioner.getBucketIndex" should "cycle through bucket indices" in { + val partitioning = RoundRobinPartitioning( + batchSize = 100, + channels = Seq(channel(r1), channel(r2), channel(r3)) + ) + val partitioner = RoundRobinPartitioner(partitioning) + + val indices = (1 to 7).map(_ => partitioner.getBucketIndex(intTuple(0)).next()).toList + // Implementation increments first, then emits. Starting from 0, the first + // emitted index is therefore 1, then 2, then 0, repeating. + assert(indices == List(1, 2, 0, 1, 2, 0, 1)) + } + + "RoundRobinPartitioner.allReceivers" should "preserve channel order while deduplicating" in { + val partitioning = RoundRobinPartitioning( + batchSize = 100, + channels = Seq(channel(r2), channel(r1), channel(r2)) + ) + val partitioner = RoundRobinPartitioner(partitioning) + assert(partitioner.allReceivers == Seq(r2, r1)) + } + + // -- HashBasedShufflePartitioner ------------------------------------------ + + "HashBasedShufflePartitioner.getBucketIndex" should "return a non-negative index within the receiver count" in { + val partitioning = HashBasedShufflePartitioning( + batchSize = 100, + channels = Seq(channel(r1), channel(r2), channel(r3)), + hashAttributeNames = Seq("k") + ) + val partitioner = HashBasedShufflePartitioner(partitioning) + + (0 until 50).foreach { i => + val idx = partitioner.getBucketIndex(stringTuple(s"key-$i", "v")).next() + assert(idx >= 0 && idx < 3, s"index $idx out of range for tuple key-$i") + } + } + + it should "be deterministic for the same hash-key value" in { + val partitioning = HashBasedShufflePartitioning( + batchSize = 100, + channels = Seq(channel(r1), channel(r2), channel(r3)), + hashAttributeNames = Seq("k") + ) + val partitioner = HashBasedShufflePartitioner(partitioning) + + val first = partitioner.getBucketIndex(stringTuple("alpha", "ignored-1")).next() + val second = partitioner.getBucketIndex(stringTuple("alpha", "ignored-2")).next() + assert(first == second) + } + + it should "depend only on the hash-attribute subset, not on other fields" in { + val partitioning = HashBasedShufflePartitioning( + batchSize = 100, + channels = Seq(channel(r1), channel(r2), channel(r3)), + hashAttributeNames = Seq("k") + ) + val partitioner = HashBasedShufflePartitioner(partitioning) + + val a = partitioner.getBucketIndex(stringTuple("same-key", "value-A")).next() + val b = partitioner.getBucketIndex(stringTuple("same-key", "value-B")).next() + assert(a == b) Review Comment: Done in b3d118031e. The test now sweeps 6 keys × 8 varying second-field values; a full-tuple-hashing implementation would have to collide modulo 3 on every single key to pass. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
