Update

Author: Sephiroth-Lin

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

@@ -213,10 +213,22 @@ private[sql] abstract class SparkStrategies extends QueryPlanner[SparkPlan] {
   object CartesianProduct extends Strategy {
     def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
       case logical.Join(left, right, _, None) =>
-        execution.joins.CartesianProduct(planLater(left), planLater(right)) :: Nil
+        val buildSide =
+          if (left.statistics.sizeInBytes <= right.statistics.sizeInBytes) {
+            joins.BuildRight
+          }  else {
+            joins.BuildLeft
+          }
+        execution.joins.CartesianProduct(planLater(left), planLater(right), buildSide) :: Nil
       case logical.Join(left, right, Inner, Some(condition)) =>
+        val buildSide =
+          if (left.statistics.sizeInBytes <= right.statistics.sizeInBytes) {
+            joins.BuildRight
+          }  else {
+            joins.BuildLeft
+          }
         execution.Filter(condition,
-          execution.joins.CartesianProduct(planLater(left), planLater(right))) :: Nil
+          execution.joins.CartesianProduct(planLater(left), planLater(right), buildSide)) :: Nil
       case _ => Nil
     }
   }

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

@@ -27,24 +27,26 @@ import org.apache.spark.sql.execution.{BinaryNode, SparkPlan}
  * :: DeveloperApi ::
  */
 @DeveloperApi
-case class CartesianProduct(left: SparkPlan, right: SparkPlan) extends BinaryNode {
-  override def output: Seq[Attribute] = left.output ++ right.output
+case class CartesianProduct(
+     left: SparkPlan,
+     right: SparkPlan,
+     buildSide: BuildSide) extends BinaryNode {
 
-  protected override def doExecute(): RDD[InternalRow] = {
-    val leftResults = left.execute().map(_.copy())
-    val rightResults = right.execute().map(_.copy())
+  private val (streamed, broadcast) = buildSide match {
+    case BuildRight => (left, right)
+    case BuildLeft => (right, left)
+  }
 
-    val cartesianRdd = if (leftResults.partitions.size > rightResults.partitions.size) {
-      rightResults.cartesian(leftResults).mapPartitions { iter =>
-        iter.map(tuple => (tuple._2, tuple._1))
-      }
-    } else {
-      leftResults.cartesian(rightResults)
-    }
+  override def output: Seq[Attribute] = left.output ++ right.output
 
-    cartesianRdd.mapPartitions { iter =>
+  protected override def doExecute(): RDD[InternalRow] = {
+    val broadcastedRelation = sparkContext.broadcast(broadcast.execute().map(_.copy()))
+    broadcastedRelation.value.cartesian(streamed.execute().map(_.copy())).mapPartitions{ iter =>
       val joinedRow = new JoinedRow
-      iter.map(r => joinedRow(r._1, r._2))
+      buildSide match {
+        case BuildRight => iter.map(r => joinedRow(r._1, r._2))
+        case BuildLeft => iter.map(r => joinedRow(r._2, r._1))
+      }
     }
   }
 }