Copilot commented on code in PR #4746: URL: https://github.com/apache/texera/pull/4746#discussion_r3177468941
########## 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: This example doesn't actually verify determinism, because the second call uses a different tuple. A partitioner with per-call state that happened to ignore `v` would still pass here, so the test doesn't lock down the 'same input -> same bucket' contract described in the PR. ########## 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: This only compares one pair of tuples with 3 buckets, so a buggy implementation that hashes the full tuple can still pass if those two full-tuple hashes collide modulo 3. As written, the test is too weak to prove that non-hash attributes are ignored. ########## 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: The assertion here never checks the advertised fallback behavior. Any implementation that returns an in-range bucket — even one that still hashes only `k` or always returns a constant index — would pass, so this test does not cover 'use the full tuple when no hash attributes are configured.' -- 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]
