fix projection

apache · Jul 22, 2015 · 6294b1e · 6294b1e
1 parent 10583f1
commit 6294b1e
Show file tree

Hide file tree

Showing 3 changed files with 15 additions and 6 deletions.
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastNestedLoopJoin.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastNestedLoopJoin.scala
@@ -47,12 +47,15 @@ case class BroadcastNestedLoopJoin(
   override def outputsUnsafeRows: Boolean = left.outputsUnsafeRows || right.outputsUnsafeRows
   override def canProcessUnsafeRows: Boolean = true
 
-  @transient private[this] lazy val resultProjection: InternalRow => InternalRow =
+  @transient private[this] lazy val resultProjection: Projection = {
     if (outputsUnsafeRows) {
       UnsafeProjection.create(schema)
     } else {
-      (r: InternalRow) => r
+      new Projection {
+        override def apply(r: InternalRow): InternalRow = r
+      }
     }
+  }
 
   override def outputPartitioning: Partitioning = streamed.outputPartitioning
 

diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashJoin.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashJoin.scala
@@ -70,12 +70,15 @@ trait HashJoin {
 
       // Mutable per row objects.
       private[this] val joinRow = new JoinedRow2
-      private[this] val resultProjection: InternalRow => InternalRow =
+      private[this] val resultProjection: Projection = {
         if (supportUnsafe) {
           UnsafeProjection.create(self.schema)
         } else {
-          (r: InternalRow) => r
+          new Projection {
+            override def apply(r: InternalRow): InternalRow = r
+          }
         }
+      }
 
       private[this] val joinKeys = streamSideKeyGenerator
 

diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashOuterJoin.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashOuterJoin.scala
@@ -92,12 +92,15 @@ trait HashOuterJoin {
     }
   }
 
-  @transient private[this] lazy val resultProjection: InternalRow => InternalRow =
+  @transient private[this] lazy val resultProjection: Projection = {
     if (supportUnsafe) {
       UnsafeProjection.create(self.schema)
     } else {
-      (r: InternalRow) => r
+      new Projection {
+        override def apply(r: InternalRow): InternalRow = r
+      }
     }
+  }
 
   @transient private[this] lazy val DUMMY_LIST = CompactBuffer[InternalRow](null)
   @transient protected[this] lazy val EMPTY_LIST = CompactBuffer[InternalRow]()