optimize code style.

Author: liutang123

sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashedRelation.scala

@@ -772,7 +772,7 @@ private[execution] final class LongToUnsafeRowMap(val mm: TaskMemoryManager, cap
     array = readLongArray(readBuffer, length)
     val pageLength = readLong().toInt
     page = readLongArray(readBuffer, pageLength)
-    // Set cursor because cursor is used in write function.
+    // Restore cursor variable to make this map able to be serialized again on executors.
     cursor = pageLength * 8 + Platform.LONG_ARRAY_OFFSET
   }
 

sql/core/src/test/scala/org/apache/spark/sql/execution/joins/HashedRelationSuite.scala

@@ -278,37 +278,34 @@ class HashedRelationSuite extends SparkFunSuite with SharedSQLContext {
     map.free()
   }
 
-  test("SPARK-24809: Serializing LongHashedRelation in executor may result in data error") {
+  test("SPARK-24809: Serializing LongToUnsafeRowMap in executor may result in data error") {
     val unsafeProj = UnsafeProjection.create(Array[DataType](LongType))
     val originalMap = new LongToUnsafeRowMap(mm, 1)
 
     val key1 = 1L
-    val value1 = new Random().nextLong()
+    val value1 = 4852306286022334418L
 
     val key2 = 2L
-    val value2 = new Random().nextLong()
+    val value2 = 8813607448788216010L
 
     originalMap.append(key1, unsafeProj(InternalRow(value1)))
     originalMap.append(key2, unsafeProj(InternalRow(value2)))
     originalMap.optimize()
 
-    val resultRow = new UnsafeRow(1)
-    assert(originalMap.getValue(key1, resultRow).getLong(0) === value1)
-    assert(originalMap.getValue(key2, resultRow).getLong(0) === value2)
-
     val ser = new KryoSerializer(
             (new SparkConf).set("spark.kryo.referenceTracking", "false")).newInstance()
+    // Simulate serialize/deserialize twice on driver and executor
+    val firstTimeSerialized = ser.deserialize[LongToUnsafeRowMap](ser.serialize(originalMap))
+    val secondTimeSerialized =
+      ser.deserialize[LongToUnsafeRowMap](ser.serialize(firstTimeSerialized))
 
-    val mapSerializedInDriver = ser.deserialize[LongToUnsafeRowMap](ser.serialize(originalMap))
-    val mapSerializedInExecutor =
-      ser.deserialize[LongToUnsafeRowMap](ser.serialize(mapSerializedInDriver))
-
-    assert(mapSerializedInExecutor.getValue(key1, resultRow).getLong(0) === value1)
-    assert(mapSerializedInExecutor.getValue(key2, resultRow).getLong(0) === value2)
+    val resultRow = new UnsafeRow(1)
+    assert(secondTimeSerialized.getValue(key1, resultRow).getLong(0) === value1)
+    assert(secondTimeSerialized.getValue(key2, resultRow).getLong(0) === value2)
 
     originalMap.free()
-    mapSerializedInDriver.free()
-    mapSerializedInExecutor.free()
+    firstTimeSerialized.free()
+    secondTimeSerialized.free()
   }
 
   test("Spark-14521") {