mridulm commented on code in PR #50230: URL: https://github.com/apache/spark/pull/50230#discussion_r2071966473
########## core/src/main/java/org/apache/spark/util/MyByteArrayOutputStream.java: ########## @@ -0,0 +1,26 @@ +/* + * 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.spark.util; + +import java.io.ByteArrayOutputStream; + +/** Subclass of ByteArrayOutputStream that exposes `buf` directly. */ +public final class MyByteArrayOutputStream extends ByteArrayOutputStream { Review Comment: Rename ? `MyByteArrayOutputStream` was fine when it was internal to the class. Something like `ExposedBufferByteArrayOutputStream` or some such ? ########## core/src/main/java/org/apache/spark/shuffle/checksum/RowBasedChecksum.scala: ########## @@ -0,0 +1,114 @@ +/* + * 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.spark.shuffle.checksum + +import java.io.ObjectOutputStream +import java.util.zip.Checksum + +import scala.util.control.NonFatal + +import org.apache.spark.internal.Logging +import org.apache.spark.network.shuffle.checksum.ShuffleChecksumHelper +import org.apache.spark.util.MyByteArrayOutputStream + +/** + * A class for computing checksum for input (key, value) pairs. The checksum is independent of + * the order of the input (key, value) pairs. It is done by computing a checksum for each row + * first, and then computing the XOR for all the row checksums. + */ +abstract class RowBasedChecksum() extends Serializable with Logging { + private var hasError: Boolean = false + private var checksumValue: Long = 0 + /** Returns the checksum value computed. Tt returns the default checksum value (0) if there + * are any errors encountered during the checksum computation. + */ + def getValue: Long = { + if (!hasError) checksumValue else 0 + } + + /** Updates the row-based checksum with the given (key, value) pair */ + def update(key: Any, value: Any): Unit = { + if (!hasError) { + try { + val rowChecksumValue = calculateRowChecksum(key, value) + checksumValue = checksumValue ^ rowChecksumValue Review Comment: Perhaps something like this might work ? https://en.wikipedia.org/wiki/Fowler%E2%80%93Noll%E2%80%93Vo_hash_function It will be more expensive than xor, but should handle order and duplication. ########## core/src/main/scala/org/apache/spark/MapOutputTracker.scala: ########## @@ -169,6 +174,12 @@ private class ShuffleStatus( } else { mapIdToMapIndex.remove(currentMapStatus.mapId) } + + val preStatus = + if (mapStatuses(mapIndex) != null) mapStatuses(mapIndex) else mapStatusesDeleted(mapIndex) + if (preStatus != null && preStatus.checksumValue != status.checksumValue) { + checksumMismatchIndices.add(mapIndex) + } Review Comment: > For case 2, if downstream stages have not consumed output, which means they have not started. In this case, the rollback is a no-op, and it doesn't hurt to record the mismatches here. It is unclear how `checksumMismatchIndices` will be used - as perhaps it might be fine to record it: but my query would be why record it at all ? Is it due to complexity of detecting case (2) ? > For case 3, I think we need to record the mismatches. Assuming a situation where all partitions of a stage have finished, while some speculative tasks are still running. As all outputs have been produced, the downstream stage can start and read from the data. Later, some speculative tasks finish, and new mapStatus will override the old mapStatus with new data location. For the downstream stage, the not yet started tasks or retried tasks would read from the new data, while the finished and running tasks would read from the old data, resulting in inconsistency. That is fair, this is indeed possible. ########## core/src/main/scala/org/apache/spark/Dependency.scala: ########## @@ -59,6 +60,9 @@ abstract class NarrowDependency[T](_rdd: RDD[T]) extends Dependency[T] { override def rdd: RDD[T] = _rdd } +object ShuffleDependency { + private val EmptyRowBasedChecksums: Array[RowBasedChecksum] = Array.empty Review Comment: ```suggestion private val EMPTY_ROW_BASED_CHECKSUMS: Array[RowBasedChecksum] = Array.empty ``` -- 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: reviews-unsubscr...@spark.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org --------------------------------------------------------------------- To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
