WIP towards copying

Author: JoshRosen

sql/catalyst/src/main/java/org/apache/spark/sql/execution/joins/SortMergeJoinIterator.java

@@ -112,7 +112,7 @@ private boolean nextMatchingPair() {
         // Iterate the right side to buffer all rows that match.
         // As the records should be ordered, exit when we meet the first record that not match.
         while (!stop && rightElement != null) {
-          rightMatches.$plus$eq(rightElement);
+          rightMatches.$plus$eq(rightElement.copy());
           fetchRight();
           stop = keyOrdering.compare(leftKey, rightKey) != 0;
         }

sql/core/src/main/scala/org/apache/spark/sql/execution/basicOperators.scala

@@ -250,8 +250,9 @@ case class ExternalSort(
 
 /**
  * :: DeveloperApi ::
- * TODO(josh): document
- * Performs a sort, spilling to disk as needed.
+ * Optimized version of [[ExternalSort]] that operates on binary data (implemented as part of
+ * Project Tungsten).
+ *
  * @param global when true performs a global sort of all partitions by shuffling the data first
  *               if necessary.
  */
@@ -262,7 +263,6 @@ case class UnsafeExternalSort(
     child: SparkPlan)
   extends UnaryNode {
 
-  private[this] val numFields: Int = child.schema.size
   private[this] val schema: StructType = child.schema
 
   override def requiredChildDistribution: Seq[Distribution] =
@@ -275,6 +275,7 @@ case class UnsafeExternalSort(
       val prefixComparator = new PrefixComparator {
         override def compare(prefix1: Long, prefix2: Long): Int = 0
       }
+      // TODO: do real prefix comparsion. For dev/testing purposes, this is a dummy implementation.
       def prefixComputer(row: Row): Long = 0
       new UnsafeExternalRowSorter(schema, ordering, prefixComparator, prefixComputer).sort(iterator)
     }

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

@@ -17,10 +17,10 @@
 
 package org.apache.spark.sql.execution.joins
 
-
 import org.apache.spark.annotation.DeveloperApi
 import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.Row
+import org.apache.spark.sql.catalyst.expressions
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.plans.physical._
 import org.apache.spark.sql.execution.{UnsafeExternalSort, BinaryNode, SparkPlan}
@@ -63,17 +63,17 @@ case class UnsafeSortMergeJoin(
 
     // Only sort if necessary:
     val leftOrder = requiredOrders(leftKeys)
-    val leftResults = {
+    val leftResults: RDD[Row] = {
       if (left.outputOrdering == leftOrder) {
-        left.execute().map(_.copy())
+        left.execute()
       } else {
         new UnsafeExternalSort(leftOrder, global = false, left).execute()
       }
     }
     val rightOrder = requiredOrders(rightKeys)
-    val rightResults = {
+    val rightResults: RDD[Row] = {
       if (right.outputOrdering == rightOrder) {
-        right.execute().map(_.copy())
+        right.execute()
       } else {
         new UnsafeExternalSort(rightOrder, global = false, right).execute()
       }