diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashJoin.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashJoin.scala
index ec1a148342fc63ae92917241fd9887b645b3e3ab..f7a68e4f5d44505b2fe9acd0870ad70a2005924c 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashJoin.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashJoin.scala
@@ -20,14 +20,14 @@ package org.apache.spark.sql.execution.joins
import scala.concurrent._
import scala.concurrent.duration._
-import org.apache.spark.{InternalAccumulator, TaskContext}
import org.apache.spark.annotation.DeveloperApi
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.Expression
import org.apache.spark.sql.catalyst.plans.physical.{Distribution, Partitioning, UnspecifiedDistribution}
-import org.apache.spark.sql.execution.{BinaryNode, SparkPlan, SQLExecution}
+import org.apache.spark.sql.execution.{BinaryNode, SQLExecution, SparkPlan}
import org.apache.spark.util.ThreadUtils
+import org.apache.spark.{InternalAccumulator, TaskContext}
/**
* :: DeveloperApi ::
@@ -102,6 +102,6 @@ case class BroadcastHashJoin(
object BroadcastHashJoin {
- private val broadcastHashJoinExecutionContext = ExecutionContext.fromExecutorService(
+ private[joins] val broadcastHashJoinExecutionContext = ExecutionContext.fromExecutorService(
ThreadUtils.newDaemonCachedThreadPool("broadcast-hash-join", 128))
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashOuterJoin.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashOuterJoin.scala
index e342fd914d3218209093149dc44d207f15dd90d7..a3626de49aeaba538d6f0d6e03b890a4cf69e175 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashOuterJoin.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashOuterJoin.scala
@@ -20,15 +20,14 @@ package org.apache.spark.sql.execution.joins
import scala.concurrent._
import scala.concurrent.duration._
-import org.apache.spark.{InternalAccumulator, TaskContext}
import org.apache.spark.annotation.DeveloperApi
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions._
-import org.apache.spark.sql.catalyst.plans.physical.{Partitioning, Distribution, UnspecifiedDistribution}
+import org.apache.spark.sql.catalyst.plans.physical.{Distribution, Partitioning, UnspecifiedDistribution}
import org.apache.spark.sql.catalyst.plans.{JoinType, LeftOuter, RightOuter}
-import org.apache.spark.sql.execution.{BinaryNode, SparkPlan, SQLExecution}
-import org.apache.spark.util.ThreadUtils
+import org.apache.spark.sql.execution.{BinaryNode, SQLExecution, SparkPlan}
+import org.apache.spark.{InternalAccumulator, TaskContext}
/**
* :: DeveloperApi ::
@@ -76,7 +75,7 @@ case class BroadcastHashOuterJoin(
val hashed = HashedRelation(input.iterator, buildKeyGenerator, input.size)
sparkContext.broadcast(hashed)
}
- }(BroadcastHashOuterJoin.broadcastHashOuterJoinExecutionContext)
+ }(BroadcastHashJoin.broadcastHashJoinExecutionContext)
}
protected override def doPrepare(): Unit = {
@@ -98,19 +97,20 @@ case class BroadcastHashOuterJoin(
case _ =>
}
+ val resultProj = resultProjection
joinType match {
case LeftOuter =>
streamedIter.flatMap(currentRow => {
val rowKey = keyGenerator(currentRow)
joinedRow.withLeft(currentRow)
- leftOuterIterator(rowKey, joinedRow, hashTable.get(rowKey))
+ leftOuterIterator(rowKey, joinedRow, hashTable.get(rowKey), resultProj)
})
case RightOuter =>
streamedIter.flatMap(currentRow => {
val rowKey = keyGenerator(currentRow)
joinedRow.withRight(currentRow)
- rightOuterIterator(rowKey, hashTable.get(rowKey), joinedRow)
+ rightOuterIterator(rowKey, hashTable.get(rowKey), joinedRow, resultProj)
})
case x =>
@@ -120,9 +120,3 @@ case class BroadcastHashOuterJoin(
}
}
}
-
-object BroadcastHashOuterJoin {
-
- private val broadcastHashOuterJoinExecutionContext = ExecutionContext.fromExecutorService(
- ThreadUtils.newDaemonCachedThreadPool("broadcast-hash-outer-join", 128))
-}
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
index 83b726a8e289704a5a4d18eeb97cbfab44fbb3a3..23aebf4b068b4bda37c36c11bdc12d20bc5a2d06 100644
--- 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,7 +47,7 @@ case class BroadcastNestedLoopJoin(
override def outputsUnsafeRows: Boolean = left.outputsUnsafeRows || right.outputsUnsafeRows
override def canProcessUnsafeRows: Boolean = true
- @transient private[this] lazy val resultProjection: InternalRow => InternalRow = {
+ private[this] def genResultProjection: InternalRow => InternalRow = {
if (outputsUnsafeRows) {
UnsafeProjection.create(schema)
} else {
@@ -88,6 +88,7 @@ case class BroadcastNestedLoopJoin(
val leftNulls = new GenericMutableRow(left.output.size)
val rightNulls = new GenericMutableRow(right.output.size)
+ val resultProj = genResultProjection
streamedIter.foreach { streamedRow =>
var i = 0
@@ -97,11 +98,11 @@ case class BroadcastNestedLoopJoin(
val broadcastedRow = broadcastedRelation.value(i)
buildSide match {
case BuildRight if boundCondition(joinedRow(streamedRow, broadcastedRow)) =>
- matchedRows += resultProjection(joinedRow(streamedRow, broadcastedRow)).copy()
+ matchedRows += resultProj(joinedRow(streamedRow, broadcastedRow)).copy()
streamRowMatched = true
includedBroadcastTuples += i
case BuildLeft if boundCondition(joinedRow(broadcastedRow, streamedRow)) =>
- matchedRows += resultProjection(joinedRow(broadcastedRow, streamedRow)).copy()
+ matchedRows += resultProj(joinedRow(broadcastedRow, streamedRow)).copy()
streamRowMatched = true
includedBroadcastTuples += i
case _ =>
@@ -111,9 +112,9 @@ case class BroadcastNestedLoopJoin(
(streamRowMatched, joinType, buildSide) match {
case (false, LeftOuter | FullOuter, BuildRight) =>
- matchedRows += resultProjection(joinedRow(streamedRow, rightNulls)).copy()
+ matchedRows += resultProj(joinedRow(streamedRow, rightNulls)).copy()
case (false, RightOuter | FullOuter, BuildLeft) =>
- matchedRows += resultProjection(joinedRow(leftNulls, streamedRow)).copy()
+ matchedRows += resultProj(joinedRow(leftNulls, streamedRow)).copy()
case _ =>
}
}
@@ -127,6 +128,8 @@ case class BroadcastNestedLoopJoin(
val leftNulls = new GenericMutableRow(left.output.size)
val rightNulls = new GenericMutableRow(right.output.size)
+ val resultProj = genResultProjection
+
/** Rows from broadcasted joined with nulls. */
val broadcastRowsWithNulls: Seq[InternalRow] = {
val buf: CompactBuffer[InternalRow] = new CompactBuffer()
@@ -138,7 +141,7 @@ case class BroadcastNestedLoopJoin(
joinedRow.withLeft(leftNulls)
while (i < rel.length) {
if (!allIncludedBroadcastTuples.contains(i)) {
- buf += resultProjection(joinedRow.withRight(rel(i))).copy()
+ buf += resultProj(joinedRow.withRight(rel(i))).copy()
}
i += 1
}
@@ -147,7 +150,7 @@ case class BroadcastNestedLoopJoin(
joinedRow.withRight(rightNulls)
while (i < rel.length) {
if (!allIncludedBroadcastTuples.contains(i)) {
- buf += resultProjection(joinedRow.withLeft(rel(i))).copy()
+ buf += resultProj(joinedRow.withLeft(rel(i))).copy()
}
i += 1
}
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
index 6b3d1652923fd495e3ae545bdff9fb1ffa431ca0..5e9cd9fd2345a89c495f0875f8d96596860462ad 100644
--- 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
@@ -52,14 +52,14 @@ trait HashJoin {
override def canProcessUnsafeRows: Boolean = isUnsafeMode
override def canProcessSafeRows: Boolean = !isUnsafeMode
- @transient protected lazy val buildSideKeyGenerator: Projection =
+ protected def buildSideKeyGenerator: Projection =
if (isUnsafeMode) {
UnsafeProjection.create(buildKeys, buildPlan.output)
} else {
newMutableProjection(buildKeys, buildPlan.output)()
}
- @transient protected lazy val streamSideKeyGenerator: Projection =
+ protected def streamSideKeyGenerator: Projection =
if (isUnsafeMode) {
UnsafeProjection.create(streamedKeys, streamedPlan.output)
} else {
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
index a323aea4ea2c40a61f604817ff489f07b6332e13..346337e64245ca8f01e2b50fc343b44085c1d950 100644
--- 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
@@ -76,14 +76,14 @@ trait HashOuterJoin {
override def canProcessUnsafeRows: Boolean = isUnsafeMode
override def canProcessSafeRows: Boolean = !isUnsafeMode
- @transient protected lazy val buildKeyGenerator: Projection =
+ protected def buildKeyGenerator: Projection =
if (isUnsafeMode) {
UnsafeProjection.create(buildKeys, buildPlan.output)
} else {
newMutableProjection(buildKeys, buildPlan.output)()
}
- @transient protected[this] lazy val streamedKeyGenerator: Projection = {
+ protected[this] def streamedKeyGenerator: Projection = {
if (isUnsafeMode) {
UnsafeProjection.create(streamedKeys, streamedPlan.output)
} else {
@@ -91,7 +91,7 @@ trait HashOuterJoin {
}
}
- @transient private[this] lazy val resultProjection: InternalRow => InternalRow = {
+ protected[this] def resultProjection: InternalRow => InternalRow = {
if (isUnsafeMode) {
UnsafeProjection.create(self.schema)
} else {
@@ -113,7 +113,8 @@ trait HashOuterJoin {
protected[this] def leftOuterIterator(
key: InternalRow,
joinedRow: JoinedRow,
- rightIter: Iterable[InternalRow]): Iterator[InternalRow] = {
+ rightIter: Iterable[InternalRow],
+ resultProjection: InternalRow => InternalRow): Iterator[InternalRow] = {
val ret: Iterable[InternalRow] = {
if (!key.anyNull) {
val temp = if (rightIter != null) {
@@ -124,12 +125,12 @@ trait HashOuterJoin {
List.empty
}
if (temp.isEmpty) {
- resultProjection(joinedRow.withRight(rightNullRow)).copy :: Nil
+ resultProjection(joinedRow.withRight(rightNullRow)) :: Nil
} else {
temp
}
} else {
- resultProjection(joinedRow.withRight(rightNullRow)).copy :: Nil
+ resultProjection(joinedRow.withRight(rightNullRow)) :: Nil
}
}
ret.iterator
@@ -138,24 +139,24 @@ trait HashOuterJoin {
protected[this] def rightOuterIterator(
key: InternalRow,
leftIter: Iterable[InternalRow],
- joinedRow: JoinedRow): Iterator[InternalRow] = {
+ joinedRow: JoinedRow,
+ resultProjection: InternalRow => InternalRow): Iterator[InternalRow] = {
val ret: Iterable[InternalRow] = {
if (!key.anyNull) {
val temp = if (leftIter != null) {
leftIter.collect {
- case l if boundCondition(joinedRow.withLeft(l)) =>
- resultProjection(joinedRow).copy()
+ case l if boundCondition(joinedRow.withLeft(l)) => resultProjection(joinedRow).copy()
}
} else {
List.empty
}
if (temp.isEmpty) {
- resultProjection(joinedRow.withLeft(leftNullRow)).copy :: Nil
+ resultProjection(joinedRow.withLeft(leftNullRow)) :: Nil
} else {
temp
}
} else {
- resultProjection(joinedRow.withLeft(leftNullRow)).copy :: Nil
+ resultProjection(joinedRow.withLeft(leftNullRow)) :: Nil
}
}
ret.iterator
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashSemiJoin.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashSemiJoin.scala
index 97fde8f975bfdd51822ef875cb1e4dba1589567e..47a7d370f5415c3ce58b89509dcb8485fa5a2fab 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashSemiJoin.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashSemiJoin.scala
@@ -43,14 +43,14 @@ trait HashSemiJoin {
override def canProcessUnsafeRows: Boolean = supportUnsafe
override def canProcessSafeRows: Boolean = !supportUnsafe
- @transient protected lazy val leftKeyGenerator: Projection =
+ protected def leftKeyGenerator: Projection =
if (supportUnsafe) {
UnsafeProjection.create(leftKeys, left.output)
} else {
newMutableProjection(leftKeys, left.output)()
}
- @transient protected lazy val rightKeyGenerator: Projection =
+ protected def rightKeyGenerator: Projection =
if (supportUnsafe) {
UnsafeProjection.create(rightKeys, right.output)
} else {
@@ -62,12 +62,11 @@ trait HashSemiJoin {
protected def buildKeyHashSet(buildIter: Iterator[InternalRow]): java.util.Set[InternalRow] = {
val hashSet = new java.util.HashSet[InternalRow]()
- var currentRow: InternalRow = null
// Create a Hash set of buildKeys
val rightKey = rightKeyGenerator
while (buildIter.hasNext) {
- currentRow = buildIter.next()
+ val currentRow = buildIter.next()
val rowKey = rightKey(currentRow)
if (!rowKey.anyNull) {
val keyExists = hashSet.contains(rowKey)
@@ -76,6 +75,7 @@ trait HashSemiJoin {
}
}
}
+
hashSet
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashedRelation.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashedRelation.scala
index 58b4236f7b5b5b6d77b7f2e0cefa021a4f9e9f39..3f257ecdd156cbe203ebd792856b8db80491e345 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashedRelation.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashedRelation.scala
@@ -17,12 +17,11 @@
package org.apache.spark.sql.execution.joins
-import java.io.{IOException, Externalizable, ObjectInput, ObjectOutput}
+import java.io.{Externalizable, IOException, ObjectInput, ObjectOutput}
import java.nio.ByteOrder
import java.util.{HashMap => JavaHashMap}
import org.apache.spark.shuffle.ShuffleMemoryManager
-import org.apache.spark.{SparkConf, SparkEnv, TaskContext}
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.execution.SparkSqlSerializer
@@ -31,6 +30,7 @@ import org.apache.spark.unsafe.map.BytesToBytesMap
import org.apache.spark.unsafe.memory.{ExecutorMemoryManager, MemoryAllocator, TaskMemoryManager}
import org.apache.spark.util.Utils
import org.apache.spark.util.collection.CompactBuffer
+import org.apache.spark.{SparkConf, SparkEnv}
/**
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/ShuffledHashOuterJoin.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/ShuffledHashOuterJoin.scala
index eee8ad800f98e0f31a7e0b2ae07ba413d35f52f2..6a8c35efca8f435f1a82a17b86aa3ebaa95fbb8b 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/ShuffledHashOuterJoin.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/ShuffledHashOuterJoin.scala
@@ -60,19 +60,21 @@ case class ShuffledHashOuterJoin(
case LeftOuter =>
val hashed = HashedRelation(rightIter, buildKeyGenerator)
val keyGenerator = streamedKeyGenerator
+ val resultProj = resultProjection
leftIter.flatMap( currentRow => {
val rowKey = keyGenerator(currentRow)
joinedRow.withLeft(currentRow)
- leftOuterIterator(rowKey, joinedRow, hashed.get(rowKey))
+ leftOuterIterator(rowKey, joinedRow, hashed.get(rowKey), resultProj)
})
case RightOuter =>
val hashed = HashedRelation(leftIter, buildKeyGenerator)
val keyGenerator = streamedKeyGenerator
+ val resultProj = resultProjection
rightIter.flatMap ( currentRow => {
val rowKey = keyGenerator(currentRow)
joinedRow.withRight(currentRow)
- rightOuterIterator(rowKey, hashed.get(rowKey), joinedRow)
+ rightOuterIterator(rowKey, hashed.get(rowKey), joinedRow, resultProj)
})
case FullOuter =>