[SPARK-36794][SQL] Ignore duplicated join keys when building relation for SEMI/ANTI shuffled hash join

### What changes were proposed in this pull request?

This is a re-proposal of #34034. Ignore duplicated join keys when building hash relation for LEFT SEMI / LEFT ANTI shuffled hash join. The original PR was also covering broadcast hash join, and caused regression for broadcast shuffle exchange reuse in #34034 (comment) . So here we do not covert broadcast hash join. Shuffled hash join do not have the shuffle reuse problem.

Make the deduplication work with broadcast exchange is kind of non-trivial for me now, because:

For two `BroadcastExchangeExec`, one with `ignoreDuplicatedKey` being true (called `exchange_A`), and the other with `ignoreDuplicatedKey` being false (called `exchange_B`):

* If we decide these two `BroadcastExchangeExec`s are not equal, then we have regression that we reuse fewer exchange after the change, as #34034 (comment) .
* If we decide these two `BroadcastExchangeExec`s are equal, then we might have correctness issue when `exchange_B` reuses result of `exchange_A`. Also these two `BroadcastExchangeExec`s should not be equal as they already have different results in underlying hash relations.

I still see it's valuable to have the optimization for shuffled hash join only. So leave the optimization for broadcast hash join as a followup as it needs more thoughts.

### Why are the changes needed?

Help reduce the hash table size of join for LEFT SEMI and LEFT ANTI, reduce OOM possibility of shuffled hash join.

### Does this PR introduce _any_ user-facing change?

No.

### How was this patch tested?

Added unit test in `JoinSuite.scala`.

Closes #34247 from c21/shj-fix.

Authored-by: Cheng Su <chengsu@fb.com>
Signed-off-by: Wenchen Fan <wenchen@databricks.com>

Author: c21

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

@@ -126,16 +126,20 @@ private[execution] object HashedRelation {
   /**
    * Create a HashedRelation from an Iterator of InternalRow.
    *
-   * @param allowsNullKey Allow NULL keys in HashedRelation.
-   *                      This is used for full outer join in `ShuffledHashJoinExec` only.
+   * @param allowsNullKey        Allow NULL keys in HashedRelation.
+   *                             This is used for full outer join in `ShuffledHashJoinExec` only.
+   * @param ignoresDuplicatedKey Ignore rows with duplicated keys in HashedRelation.
+   *                             This is only used for semi and anti join without join condition in
+   *                             `ShuffledHashJoinExec` only.
    */
   def apply(
       input: Iterator[InternalRow],
       key: Seq[Expression],
       sizeEstimate: Int = 64,
       taskMemoryManager: TaskMemoryManager = null,
       isNullAware: Boolean = false,
-      allowsNullKey: Boolean = false): HashedRelation = {
+      allowsNullKey: Boolean = false,
+      ignoresDuplicatedKey: Boolean = false): HashedRelation = {
     val mm = Option(taskMemoryManager).getOrElse {
       new TaskMemoryManager(
         new UnifiedMemoryManager(
@@ -152,7 +156,8 @@ private[execution] object HashedRelation {
       // NOTE: LongHashedRelation does not support NULL keys.
       LongHashedRelation(input, key, sizeEstimate, mm, isNullAware)
     } else {
-      UnsafeHashedRelation(input, key, sizeEstimate, mm, isNullAware, allowsNullKey)
+      UnsafeHashedRelation(input, key, sizeEstimate, mm, isNullAware, allowsNullKey,
+        ignoresDuplicatedKey)
     }
   }
 }
@@ -450,7 +455,8 @@ private[joins] object UnsafeHashedRelation {
       sizeEstimate: Int,
       taskMemoryManager: TaskMemoryManager,
       isNullAware: Boolean = false,
-      allowsNullKey: Boolean = false): HashedRelation = {
+      allowsNullKey: Boolean = false,
+      ignoresDuplicatedKey: Boolean = false): HashedRelation = {
     require(!(isNullAware && allowsNullKey),
       "isNullAware and allowsNullKey cannot be enabled at same time")
 
@@ -471,12 +477,14 @@ private[joins] object UnsafeHashedRelation {
       val key = keyGenerator(row)
       if (!key.anyNull || allowsNullKey) {
         val loc = binaryMap.lookup(key.getBaseObject, key.getBaseOffset, key.getSizeInBytes)
-        val success = loc.append(
-          key.getBaseObject, key.getBaseOffset, key.getSizeInBytes,
-          row.getBaseObject, row.getBaseOffset, row.getSizeInBytes)
-        if (!success) {
-          binaryMap.free()
-          throw QueryExecutionErrors.cannotAcquireMemoryToBuildUnsafeHashedRelationError()
+        if (!(ignoresDuplicatedKey && loc.isDefined)) {
+          val success = loc.append(
+            key.getBaseObject, key.getBaseOffset, key.getSizeInBytes,
+            row.getBaseObject, row.getBaseOffset, row.getSizeInBytes)
+          if (!success) {
+            binaryMap.free()
+            throw QueryExecutionErrors.cannotAcquireMemoryToBuildUnsafeHashedRelationError()
+          }
         }
       } else if (isNullAware) {
         binaryMap.free()

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

@@ -59,6 +59,17 @@ case class ShuffledHashJoinExec(
     case _ => super.outputOrdering
   }
 
+  // Exposed for testing
+  @transient lazy val ignoreDuplicatedKey = joinType match {
+    case LeftExistence(_) =>
+      // For building hash relation, ignore duplicated rows with same join keys if:
+      // 1. Join condition is empty, or
+      // 2. Join condition only references streamed attributes and build join keys.
+      val streamedOutputAndBuildKeys = AttributeSet(streamedOutput ++ buildKeys)
+      condition.forall(_.references.subsetOf(streamedOutputAndBuildKeys))
+    case _ => false
+  }
+
   /**
    * This is called by generated Java class, should be public.
    */
@@ -72,7 +83,8 @@ case class ShuffledHashJoinExec(
       buildBoundKeys,
       taskMemoryManager = context.taskMemoryManager(),
       // Full outer join needs support for NULL key in HashedRelation.
-      allowsNullKey = joinType == FullOuter)
+      allowsNullKey = joinType == FullOuter,
+      ignoresDuplicatedKey = ignoreDuplicatedKey)
     buildTime += NANOSECONDS.toMillis(System.nanoTime() - start)
     buildDataSize += relation.estimatedSize
     // This relation is usually used until the end of task.

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

@@ -1402,4 +1402,42 @@ class JoinSuite extends QueryTest with SharedSparkSession with AdaptiveSparkPlan
       assertJoin(sql, classOf[ShuffledHashJoinExec])
     }
   }
+
+  test("SPARK-36794: Ignore duplicated key when building relation for semi/anti hash join") {
+    withTable("t1", "t2") {
+      spark.range(10).map(i => (i.toString, i + 1)).toDF("c1", "c2").write.saveAsTable("t1")
+      spark.range(10).map(i => ((i % 5).toString, i % 3)).toDF("c1", "c2").write.saveAsTable("t2")
+
+      val semiJoinQueries = Seq(
+        // No join condition, ignore duplicated key.
+        (s"SELECT /*+ SHUFFLE_HASH(t2) */ t1.c1 FROM t1 LEFT SEMI JOIN t2 ON t1.c1 = t2.c1",
+          true),
+        // Have join condition on build join key only, ignore duplicated key.
+        (s"""
+            |SELECT /*+ SHUFFLE_HASH(t2) */ t1.c1 FROM t1 LEFT SEMI JOIN t2
+            |ON t1.c1 = t2.c1 AND CAST(t1.c2 * 2 AS STRING) != t2.c1
+          """.stripMargin,
+          true),
+        // Have join condition on other build attribute beside join key, do not ignore
+        // duplicated key.
+        (s"""
+            |SELECT /*+ SHUFFLE_HASH(t2) */ t1.c1 FROM t1 LEFT SEMI JOIN t2
+            |ON t1.c1 = t2.c1 AND t1.c2 * 100 != t2.c2
+          """.stripMargin,
+          false)
+      )
+      semiJoinQueries.foreach {
+        case (query, ignoreDuplicatedKey) =>
+          val semiJoinDF = sql(query)
+          val antiJoinDF = sql(query.replaceAll("SEMI", "ANTI"))
+          checkAnswer(semiJoinDF, Seq(Row("0"), Row("1"), Row("2"), Row("3"), Row("4")))
+          checkAnswer(antiJoinDF, Seq(Row("5"), Row("6"), Row("7"), Row("8"), Row("9")))
+          Seq(semiJoinDF, antiJoinDF).foreach { df =>
+            assert(collect(df.queryExecution.executedPlan) {
+              case j: ShuffledHashJoinExec if j.ignoreDuplicatedKey == ignoreDuplicatedKey => true
+            }.size == 1)
+          }
+      }
+    }
+  }
 }