rename classes

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/continuous/shuffle/ContinuousShuffleReadRDD.scala

@@ -34,7 +34,8 @@ case class ContinuousShuffleReadPartition(
   // Initialized only on the executor, and only once even as we call compute() multiple times.
   lazy val (reader: ContinuousShuffleReader, endpoint) = {
     val env = SparkEnv.get.rpcEnv
-    val receiver = new UnsafeRowReceiver(queueSize, numShuffleWriters, epochIntervalMs, env)
+    val receiver = new RPCContinuousShuffleReader(
+      queueSize, numShuffleWriters, epochIntervalMs, env)
     val endpoint = env.setupEndpoint(s"UnsafeRowReceiver-${UUID.randomUUID()}", receiver)
 
     TaskContext.get().addTaskCompletionListener { ctx =>

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/continuous/shuffle/UnsafeRowReceiver.scala → sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/continuous/shuffle/RPCContinuousShuffleReader.scala

@@ -46,7 +46,7 @@ private[shuffle] case class ReceiverEpochMarker(writerId: Int) extends UnsafeRow
  * TODO: Support multiple source tasks. We need to output a single epoch marker once all
  * source tasks have sent one.
  */
-private[shuffle] class UnsafeRowReceiver(
+private[shuffle] class RPCContinuousShuffleReader(
       queueSize: Int,
       numShuffleWriters: Int,
       epochIntervalMs: Long,

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/continuous/shuffle/UnsafeRowWriter.scala → sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/continuous/shuffle/RPCContinuousShuffleWriter.scala

@@ -22,14 +22,14 @@ import org.apache.spark.rpc.RpcEndpointRef
 import org.apache.spark.sql.catalyst.expressions.UnsafeRow
 
 /**
- * A [[ContinuousShuffleWriter]] sending data to [[UnsafeRowReceiver]] instances.
+ * A [[ContinuousShuffleWriter]] sending data to [[RPCContinuousShuffleReader]] instances.
  *
- * @param writerId The partition ID of this writer.
+ * @param writerId          The partition ID of this writer.
  * @param outputPartitioner The partitioner on the reader side of the shuffle.
- * @param endpoints The [[UnsafeRowReceiver]] endpoints to write to. Indexed by partition ID within
- *                  outputPartitioner.
+ * @param endpoints         The [[RPCContinuousShuffleReader]] endpoints to write to. Indexed by
+ *                          partition ID within outputPartitioner.
  */
-class UnsafeRowWriter(
+class RPCContinuousShuffleWriter(
     writerId: Int,
     outputPartitioner: Partitioner,
     endpoints: Array[RpcEndpointRef]) extends ContinuousShuffleWriter {

sql/core/src/test/scala/org/apache/spark/sql/streaming/continuous/shuffle/ContinuousShuffleReadSuite.scala

@@ -70,7 +70,7 @@ class ContinuousShuffleReadSuite extends StreamTest {
     ctx.markTaskCompleted(None)
     val receiver = rdd.partitions(0).asInstanceOf[ContinuousShuffleReadPartition].reader
     eventually(timeout(streamingTimeout)) {
-      assert(receiver.asInstanceOf[UnsafeRowReceiver].stopped.get())
+      assert(receiver.asInstanceOf[RPCContinuousShuffleReader].stopped.get())
     }
   }
 
@@ -86,7 +86,7 @@ class ContinuousShuffleReadSuite extends StreamTest {
     ctx.markTaskCompleted(None)
     val receiver = rdd.partitions(0).asInstanceOf[ContinuousShuffleReadPartition].reader
     eventually(timeout(streamingTimeout)) {
-      assert(receiver.asInstanceOf[UnsafeRowReceiver].stopped.get())
+      assert(receiver.asInstanceOf[RPCContinuousShuffleReader].stopped.get())
     }
   }
 

sql/core/src/test/scala/org/apache/spark/sql/streaming/continuous/shuffle/ContinuousShuffleSuite.scala

@@ -22,7 +22,7 @@ import scala.collection.mutable
 import org.apache.spark.{HashPartitioner, Partition, TaskContext, TaskContextImpl}
 import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.catalyst.expressions.{GenericInternalRow, UnsafeProjection, UnsafeRow}
-import org.apache.spark.sql.execution.streaming.continuous.shuffle.{ContinuousShuffleReadPartition, ContinuousShuffleReadRDD, UnsafeRowWriter}
+import org.apache.spark.sql.execution.streaming.continuous.shuffle.{ContinuousShuffleReadPartition, ContinuousShuffleReadRDD, RPCContinuousShuffleWriter}
 import org.apache.spark.sql.streaming.StreamTest
 import org.apache.spark.sql.types.{DataType, IntegerType}
 
@@ -85,7 +85,8 @@ class ContinuousShuffleSuite extends StreamTest {
 
   test("one epoch") {
     val reader = new ContinuousShuffleReadRDD(sparkContext, numPartitions = 1)
-    val writer = new UnsafeRowWriter(0, new HashPartitioner(1), Array(readRDDEndpoint(reader)))
+    val writer = new RPCContinuousShuffleWriter(
+      0, new HashPartitioner(1), Array(readRDDEndpoint(reader)))
 
     writer.write(Iterator(1, 2, 3))
 
@@ -94,7 +95,8 @@ class ContinuousShuffleSuite extends StreamTest {
 
   test("multiple epochs") {
     val reader = new ContinuousShuffleReadRDD(sparkContext, numPartitions = 1)
-    val writer = new UnsafeRowWriter(0, new HashPartitioner(1), Array(readRDDEndpoint(reader)))
+    val writer = new RPCContinuousShuffleWriter(
+      0, new HashPartitioner(1), Array(readRDDEndpoint(reader)))
 
     writer.write(Iterator(1, 2, 3))
     writer.write(Iterator(4, 5, 6))
@@ -107,7 +109,8 @@ class ContinuousShuffleSuite extends StreamTest {
     val data = new MultipleEpochRDD(1, Array(), Array(1, 2), Array(), Array(), Array(3, 4), Array())
 
     val reader = new ContinuousShuffleReadRDD(sparkContext, numPartitions = 1)
-    val writer = new UnsafeRowWriter(0, new HashPartitioner(1), Array(readRDDEndpoint(reader)))
+    val writer = new RPCContinuousShuffleWriter(
+      0, new HashPartitioner(1), Array(readRDDEndpoint(reader)))
 
     writer.write(Iterator())
     writer.write(Iterator(1, 2))
@@ -126,7 +129,8 @@ class ContinuousShuffleSuite extends StreamTest {
 
   test("blocks waiting for writer") {
     val reader = new ContinuousShuffleReadRDD(sparkContext, numPartitions = 1)
-    val writer = new UnsafeRowWriter(0, new HashPartitioner(1), Array(readRDDEndpoint(reader)))
+    val writer = new RPCContinuousShuffleWriter(
+      0, new HashPartitioner(1), Array(readRDDEndpoint(reader)))
 
     val readerEpoch = reader.compute(reader.partitions(0), ctx)
 
@@ -152,7 +156,7 @@ class ContinuousShuffleSuite extends StreamTest {
     val reader = new ContinuousShuffleReadRDD(
       sparkContext, numPartitions = 1, numShuffleWriters = numWriterPartitions)
     val writers = (0 until 3).map { idx =>
-      new UnsafeRowWriter(idx, new HashPartitioner(1), Array(readRDDEndpoint(reader)))
+      new RPCContinuousShuffleWriter(idx, new HashPartitioner(1), Array(readRDDEndpoint(reader)))
     }
 
     writers(0).write(Iterator(1, 4, 7))
@@ -176,7 +180,7 @@ class ContinuousShuffleSuite extends StreamTest {
     val reader = new ContinuousShuffleReadRDD(
       sparkContext, numPartitions = 1, numShuffleWriters = numWriterPartitions)
     val writers = (0 until 3).map { idx =>
-      new UnsafeRowWriter(idx, new HashPartitioner(1), Array(readRDDEndpoint(reader)))
+      new RPCContinuousShuffleWriter(idx, new HashPartitioner(1), Array(readRDDEndpoint(reader)))
     }
 
     writers(1).write(Iterator())