update

core/src/main/scala/org/apache/spark/util/collection/OpenHashSet.scala

@@ -122,8 +122,8 @@ class OpenHashSet[@specialized(Long, Int, Double, Float) T: ClassTag](
    * Remove an element from the set. If an element does not exists in the set, nothing is done.
    */
   def remove(k: T): Unit = {
-    val pos = hashcode(hasher.hash(k)) & _mask
-    if (_bitset.get(pos)) {
+    val pos = getPos(k)
+    if (pos != INVALID_POS && _bitset.get(pos)) {
       _bitset.unset(pos)
       _size -= 1
     }

core/src/test/scala/org/apache/spark/util/collection/OpenHashSetSuite.scala

@@ -74,25 +74,44 @@ class OpenHashSetSuite extends SparkFunSuite with Matchers {
     assert(set.contains(999))
     assert(!set.contains(10000))
 
+    set.add(1132)  // Cause hash contention with 999
+    assert(set.size === 4)
+    assert(set.contains(10))
+    assert(set.contains(50))
+    assert(set.contains(999))
+    assert(set.contains(1132))
+    assert(!set.contains(10000))
+
+    set.remove(1132)
+    assert(set.size === 3)
+    assert(set.contains(10))
+    assert(set.contains(50))
+    assert(set.contains(999))
+    assert(!set.contains(1132))
+    assert(!set.contains(10000))
+
     set.remove(999)
     assert(set.size === 2)
     assert(set.contains(10))
     assert(set.contains(50))
     assert(!set.contains(999))
+    assert(!set.contains(1132))
     assert(!set.contains(10000))
 
-    set.add(999)
+    set.add(1132)
     assert(set.size === 3)
     assert(set.contains(10))
     assert(set.contains(50))
-    assert(set.contains(999))
+    assert(!set.contains(999))
+    assert(set.contains(1132))
     assert(!set.contains(10000))
 
     set.remove(10000)
     assert(set.size === 3)
     assert(set.contains(10))
     assert(set.contains(50))
-    assert(set.contains(999))
+    assert(!set.contains(999))
+    assert(set.contains(1132))
     assert(!set.contains(10000))
   }
 
@@ -132,25 +151,44 @@ class OpenHashSetSuite extends SparkFunSuite with Matchers {
     assert(set.contains(999L))
     assert(!set.contains(10000L))
 
+    set.add(1132L)  // Cause hash contention with 999L
+    assert(set.size === 4)
+    assert(set.contains(10L))
+    assert(set.contains(50L))
+    assert(set.contains(999L))
+    assert(set.contains(1132L))
+    assert(!set.contains(10000L))
+
+    set.remove(1132)
+    assert(set.size === 3)
+    assert(set.contains(10L))
+    assert(set.contains(50L))
+    assert(set.contains(999L))
+    assert(!set.contains(1132L))
+    assert(!set.contains(10000L))
+
     set.remove(999L)
     assert(set.size === 2)
     assert(set.contains(10L))
     assert(set.contains(50L))
     assert(!set.contains(999L))
+    assert(!set.contains(1132L))
     assert(!set.contains(10000L))
 
-    set.add(999L)
+    set.add(1132L)
     assert(set.size === 3)
     assert(set.contains(10L))
     assert(set.contains(50L))
-    assert(set.contains(999L))
+    assert(!set.contains(999L))
+    assert(set.contains(1132L))
     assert(!set.contains(10000L))
 
     set.remove(10000L)
     assert(set.size === 3)
     assert(set.contains(10L))
     assert(set.contains(50L))
-    assert(set.contains(999L))
+    assert(!set.contains(999L))
+    assert(set.contains(1132L))
     assert(!set.contains(10000L))
   }
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/collectionOperations.scala

@@ -3980,6 +3980,10 @@ object ArrayUnion {
   """,
   since = "2.4.0")
 case class ArrayIntersect(left: Expression, right: Expression) extends ArraySetLike {
+  override def dataType: DataType = ArrayType(elementType,
+    left.dataType.asInstanceOf[ArrayType].containsNull &&
+      right.dataType.asInstanceOf[ArrayType].containsNull)
+
   var hsInt: OpenHashSet[Int] = _
   var hsLong: OpenHashSet[Long] = _
 

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CollectionExpressionsSuite.scala

@@ -1654,6 +1654,7 @@ class CollectionExpressionsSuite extends SparkFunSuite with ExpressionEvalHelper
     checkEvaluation(ArrayIntersect(a01, a00), Seq(2, 4))
     checkEvaluation(ArrayIntersect(a02, a03), Seq(4, 2))
     checkEvaluation(ArrayIntersect(a03, a02), Seq(2, 4))
+    checkEvaluation(ArrayIntersect(a00, a04), Seq(1, 2, 4))
     checkEvaluation(ArrayIntersect(a04, a05), Seq(4, null, 2))
     checkEvaluation(ArrayIntersect(a02, a06), Seq.empty)
     checkEvaluation(ArrayIntersect(a06, a04), Seq.empty)
@@ -1715,8 +1716,9 @@ class CollectionExpressionsSuite extends SparkFunSuite with ExpressionEvalHelper
     checkEvaluation(ArrayIntersect(aa1, aa0), Seq[Seq[Int]](Seq[Int](3, 4)))
 
     assert(ArrayIntersect(a00, a01).dataType.asInstanceOf[ArrayType].containsNull === false)
-    assert(ArrayIntersect(a00, a04).dataType.asInstanceOf[ArrayType].containsNull === true)
+    assert(ArrayIntersect(a00, a04).dataType.asInstanceOf[ArrayType].containsNull === false)
+    assert(ArrayIntersect(a04, a05).dataType.asInstanceOf[ArrayType].containsNull === true)
     assert(ArrayIntersect(a20, a21).dataType.asInstanceOf[ArrayType].containsNull === false)
-    assert(ArrayIntersect(a20, a23).dataType.asInstanceOf[ArrayType].containsNull === true)
+    assert(ArrayIntersect(a23, a24).dataType.asInstanceOf[ArrayType].containsNull === true)
   }
 }