Spark SQL uses broadcast join (aka broadcast hash join) instead of hash join to optimize join queries when the size of one side data is below spark.sql.autoBroadcastJoinThreshold.

Broadcast join can be very efficient for joins between a large table (fact) with relatively small tables (dimensions) that could then be used to perform a star-schema join. It can avoid sending all data of the large table over the network.

testTable3= testTable1.join(broadcast(testTable2), Seq("id"), "right_outer")

    private def canBroadcast(plan: LogicalPlan): Boolean = {plan.statistics.isBroadcastable ||(plan.statistics.sizeInBytes >= 0 &&plan.statistics.sizeInBytes <= conf.autoBroadcastJoinThreshold)}

  override protected[sql] def doExecuteBroadcast[T](): broadcast.Broadcast[T] = {ThreadUtils.awaitResultInForkJoinSafely(relationFuture, timeout).asInstanceOf[broadcast.Broadcast[T]]}

  private lazy val relationFuture: Future[broadcast.Broadcast[Any]] = {// broadcastFuture is used in "doExecute". Therefore we can get the execution id correctly here.
val executionId = sparkContext.getLocalProperty(SQLExecution.EXECUTION_ID_KEY)Future {// This will run in another thread. Set the execution id so that we can connect these jobs// with the correct execution.
SQLExecution.withExecutionId(sparkContext, executionId) {try {val beforeCollect = System.nanoTime()// Note that we use .executeCollect() because we don't want to convert data to Scala types
val input: Array[InternalRow] = child.executeCollect()if (input.length >= 512000000) {throw new SparkException(s"Cannot broadcast the table with more than 512 millions rows: ${input.length} rows")}val beforeBuild = System.nanoTime()longMetric("collectTime") += (beforeBuild - beforeCollect) / 1000000val dataSize = input.map(_.asInstanceOf[UnsafeRow].getSizeInBytes.toLong).sumlongMetric("dataSize") += dataSizeif (dataSize >= (8L << 30)) {throw new SparkException(s"Cannot broadcast the table that is larger than 8GB: ${dataSize >> 30} GB")}// Construct and broadcast the relation.
val relation = mode.transform(input)val beforeBroadcast = System.nanoTime()longMetric("buildTime") += (beforeBroadcast - beforeBuild) / 1000000val broadcasted = sparkContext.broadcast(relation)longMetric("broadcastTime") += (System.nanoTime() - beforeBroadcast) / 1000000SQLMetrics.postDriverMetricUpdates(sparkContext, executionId, metrics.values.toSeq)broadcasted