mridulm commented on code in PR #2456:
URL: https://github.com/apache/celeborn/pull/2456#discussion_r1591992361
##########
common/src/main/scala/org/apache/celeborn/common/util/PbSerDeUtils.scala:
##########
@@ -490,4 +490,170 @@ object PbSerDeUtils {
pbWorkerEventInfo.getWorkerEventType.getNumber,
pbWorkerEventInfo.getEventStartTime())
}
+
+ private def toPackedPartitionLocation(
+ pbPackedLocationsBuilder: PbPackedPartitionLocations.Builder,
+ workerIdIndex: Map[String, Int],
+ mountPointsIndex: Map[String, Int],
+ location: PartitionLocation): PbPackedPartitionLocations.Builder = {
+ pbPackedLocationsBuilder.addIds(location.getId)
+ pbPackedLocationsBuilder.addEpoches(location.getEpoch)
+
pbPackedLocationsBuilder.addWorkerIds(workerIdIndex(location.getWorker.toUniqueId()))
+ pbPackedLocationsBuilder.addMapIdBitMap(
+ Utils.roaringBitmapToByteString(location.getMapIdBitMap))
+ pbPackedLocationsBuilder.addTypes(location.getStorageInfo.getType.getValue)
+ pbPackedLocationsBuilder.addMountPoints(
+ mountPointsIndex(location.getStorageInfo.getMountPoint))
+
pbPackedLocationsBuilder.addFinalResult(location.getStorageInfo.isFinalResult)
+ if (location.getStorageInfo.getFilePath != null &&
location.getStorageInfo.getFilePath.nonEmpty) {
+ pbPackedLocationsBuilder.addFilePaths(location.getStorageInfo.getFilePath
+ .substring(location.getStorageInfo.getMountPoint.length))
+ } else {
+ pbPackedLocationsBuilder.addFilePaths("")
+ }
+
pbPackedLocationsBuilder.addAvailableStorageTypes(location.getStorageInfo.availableStorageTypes)
+ pbPackedLocationsBuilder.addModes(location.getMode.mode())
+ }
+
+ def toPbPackedPartitionLocationsPair(inputLocations: List[PartitionLocation])
+ : PbPackedPartitionLocationsPair = {
+ val packedLocationPairsBuilder =
PbPackedPartitionLocationsPair.newBuilder()
+ val packedLocationsBuilder = PbPackedPartitionLocations.newBuilder()
+
+ val implicateLocations = inputLocations.map(_.getPeer).filterNot(_ == null)
+
+ val allLocations = (inputLocations ++ implicateLocations)
+ val workerIdList = new util.ArrayList[String](
+ allLocations.map(_.getWorker.toUniqueId()).toSet.asJava)
+ val workerIdIndex = workerIdList.asScala.zipWithIndex.toMap
+ val mountPointsList = new util.ArrayList[String](
+ allLocations.map(
+ _.getStorageInfo.getMountPoint).toSet.asJava)
+ val mountPointsIndex = mountPointsList.asScala.zipWithIndex.toMap
+
+ packedLocationsBuilder.addAllWorkerIdsSet(workerIdList)
+ packedLocationsBuilder.addAllMountPointsSet(mountPointsList)
+
+ val locationIndexes = allLocations.map(_.hashCode()).zipWithIndex.toMap
+
+ for (location <- allLocations) {
+ toPackedPartitionLocation(
+ packedLocationsBuilder,
+ workerIdIndex,
+ mountPointsIndex,
+ location)
+ if (location.getPeer != null) {
+ packedLocationPairsBuilder.addPeerIndexes(
+ locationIndexes(location.getPeer.hashCode()))
+ } else {
+ packedLocationPairsBuilder.addPeerIndexes(-1)
Review Comment:
`PartitionLocation.hashCode` can be `-1` - and yet be a valid peer.
If we want to make the assumption that `-1` means no peer, we should enforce
it in `PartitionLocation.hashCode`.
For example, change it to something like:
```
public final int hashCode() {
int code = (id + epoch + host + rpcPort + pushPort +
fetchPort).hashCode();
// assuming MISSING_PEER_ID == -1
return MISSING_PEER_ID == code ? 0 : code;
}
```
--
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]
