Address all new comments

Author: c21

core/src/main/java/org/apache/spark/unsafe/map/BytesToBytesMap.java

@@ -429,7 +429,7 @@ public MapIterator destructiveIterator() {
   }
 
   /**
-   * Iterator for the entries of this map. This is to first iterate over key index array
+   * Iterator for the entries of this map. This is to first iterate over key indices in
    * `longArray` then accessing values in `dataPages`. NOTE: this is different from `MapIterator`
    * in the sense that key index is preserved here
    * (See `UnsafeHashedRelation` for example of usage).

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -329,7 +329,9 @@ object SQLConf {
 
   val PREFER_SORTMERGEJOIN = buildConf("spark.sql.join.preferSortMergeJoin")
     .internal()
-    .doc("When true, prefer sort merge join over shuffle hash join.")
+    .doc("When true, prefer sort merge join over shuffled hash join. " +
+      "Note that shuffled hash join supports all join types (e.g. full outer) " +
+      "that sort merge join supports.")
     .version("2.0.0")
     .booleanConf
     .createWithDefault(true)

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

@@ -137,7 +137,7 @@ private[execution] object HashedRelation {
         0)
     }
 
-    if (!input.hasNext) {
+    if (!input.hasNext && !allowsNullKey) {
       EmptyHashedRelation
     } else if (key.length == 1 && key.head.dataType == LongType && !allowsNullKey) {
       // NOTE: LongHashedRelation does not support NULL keys.

sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala

@@ -1199,6 +1199,18 @@ class JoinSuite extends QueryTest with SharedSparkSession with AdaptiveSparkPlan
       (spark.range(10).selectExpr("id % 5 as k1"),
         spark.range(30).selectExpr("id % 5 as k2"),
         $"k1" === $"k2"),
+      // Test empty build side
+      (spark.range(10).selectExpr("id as k1").filter("k1 < -1"),
+        spark.range(30).selectExpr("id as k2"),
+        $"k1" === $"k2"),
+      // Test empty stream side
+      (spark.range(10).selectExpr("id as k1"),
+        spark.range(30).selectExpr("id as k2").filter("k2 < -1"),
+        $"k1" === $"k2"),
+      // Test empty build and stream side
+      (spark.range(10).selectExpr("id as k1").filter("k1 < -1"),
+        spark.range(30).selectExpr("id as k2").filter("k2 < -1"),
+        $"k1" === $"k2"),
       // Test string join key
       (spark.range(10).selectExpr("cast(id * 3 as string) as k1"),
         spark.range(30).selectExpr("cast(id as string) as k2"),