handle broadcast side is empty and add a test to cover it

Author: linzebing

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

@@ -461,7 +461,7 @@ case class BroadcastNestedLoopJoinExec(
   }
 
   private def codegenOuter(ctx: CodegenContext, input: Seq[ExprCode]): String = {
-    val (_, buildRowArrayTerm) = prepareBroadcast(ctx)
+    val (buildRowArray, buildRowArrayTerm) = prepareBroadcast(ctx)
     val (buildRow, checkCondition, _) = getJoinCondition(ctx, input, streamed, broadcast)
     val buildVars = genBuildSideVars(ctx, buildRow, broadcast)
 
@@ -474,25 +474,33 @@ case class BroadcastNestedLoopJoinExec(
     val foundMatch = ctx.freshName("foundMatch")
     val numOutput = metricTerm(ctx, "numOutputRows")
 
-    s"""
-       |boolean $foundMatch = false;
-       |for (int $arrayIndex = 0; $arrayIndex < $buildRowArrayTerm.length; $arrayIndex++) {
-       |  UnsafeRow $buildRow = (UnsafeRow) $buildRowArrayTerm[$arrayIndex];
-       |  boolean $shouldOutputRow = false;
-       |  $checkCondition {
-       |    $shouldOutputRow = true;
-       |    $foundMatch = true;
-       |  }
-       |  if ($arrayIndex == $buildRowArrayTerm.length - 1 && !$foundMatch) {
-       |    $buildRow = null;
-       |    $shouldOutputRow = true;
-       |  }
-       |  if ($shouldOutputRow) {
-       |    $numOutput.add(1);
-       |    ${consume(ctx, resultVars)}
-       |  }
-       |}
-     """.stripMargin
+    if (buildRowArray.isEmpty) {
+      s"""
+         |UnsafeRow $buildRow = null;
+         |$numOutput.add(1);
+         |${consume(ctx, resultVars)}
+       """.stripMargin
+    } else {
+      s"""
+         |boolean $foundMatch = false;
+         |for (int $arrayIndex = 0; $arrayIndex < $buildRowArrayTerm.length; $arrayIndex++) {
+         |  UnsafeRow $buildRow = (UnsafeRow) $buildRowArrayTerm[$arrayIndex];
+         |  boolean $shouldOutputRow = false;
+         |  $checkCondition {
+         |    $shouldOutputRow = true;
+         |    $foundMatch = true;
+         |  }
+         |  if ($arrayIndex == $buildRowArrayTerm.length - 1 && !$foundMatch) {
+         |    $buildRow = null;
+         |    $shouldOutputRow = true;
+         |  }
+         |  if ($shouldOutputRow) {
+         |    $numOutput.add(1);
+         |    ${consume(ctx, resultVars)}
+         |  }
+         |}
+       """.stripMargin
+    }
   }
 
   private def codegenLeftExistence(

sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala

@@ -215,6 +215,7 @@ class WholeStageCodegenSuite extends QueryTest with SharedSparkSession
     val df1 = spark.range(4).select($"id".as("k1"))
     val df2 = spark.range(3).select($"id".as("k2"))
     val df3 = spark.range(2).select($"id".as("k3"))
+    val df4 = spark.range(0).select($"id".as("k4"))
 
     Seq(true, false).foreach { codegenEnabled =>
       withSQLConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> codegenEnabled.toString) {
@@ -240,11 +241,19 @@ class WholeStageCodegenSuite extends QueryTest with SharedSparkSession
           .join(df3, $"k1" <= $"k3", "left_outer")
         hasJoinInCodegen = twoJoinsDF.queryExecution.executedPlan.collect {
           case WholeStageCodegenExec(BroadcastNestedLoopJoinExec(
-          _: BroadcastNestedLoopJoinExec, _, _, _, _)) => true
+            _: BroadcastNestedLoopJoinExec, _, _, _, _)) => true
         }.size === 1
         assert(hasJoinInCodegen == codegenEnabled)
         checkAnswer(twoJoinsDF,
           Seq(Row(2, 0, null), Row(3, 0, null), Row(3, 1, null), Row(null, 2, null)))
+
+        // test build side is empty
+        val buildSideIsEmptyDF = df3.join(df4, $"k3" > $"k4", "left_outer")
+        hasJoinInCodegen = buildSideIsEmptyDF.queryExecution.executedPlan.collect {
+          case WholeStageCodegenExec(_: BroadcastNestedLoopJoinExec) => true
+        }.size === 1
+        assert(hasJoinInCodegen == codegenEnabled)
+        checkAnswer(buildSideIsEmptyDF, Seq(Row(0, null), Row(1, null)))
       }
     }
   }