SPARK-49618. Fixed a bug in equality code and added bug test

apache · Sep 14, 2024 · 5c02cb3 · 5c02cb3
1 parent 1ee1b1c
commit 5c02cb3
Show file tree

Hide file tree

Showing 2 changed files with 43 additions and 3 deletions.
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/UnionEquality.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/UnionEquality.scala
@@ -17,6 +17,8 @@
 
 package org.apache.spark.sql.catalyst.plans
 
+import scala.collection.mutable
+
 import org.apache.spark.sql.catalyst.expressions.{Attribute, ExprId}
 
 trait UnionEquality [PlanType <: QueryPlan[PlanType]] {
@@ -28,9 +30,22 @@ trait UnionEquality [PlanType <: QueryPlan[PlanType]] {
   def positionAgnosticEquals(that: PlanType): Boolean = {
     val thisChildren = this.children
     val thatChildren = that.children
-    thisChildren.length == thatChildren.length &&
-      normalizeOutputAttributes(this.asInstanceOf[PlanType]) == normalizeOutputAttributes(that) &&
-      thisChildren.toSet == thatChildren.toSet
+    if (thisChildren.length == thatChildren.length &&
+      normalizeOutputAttributes(this.asInstanceOf[PlanType]) == normalizeOutputAttributes(that)) {
+      // check for elements equality disregarding order
+      val buff = mutable.ArrayBuffer[PlanType](thatChildren: _*)
+      thisChildren.forall(p => {
+        val i = buff.indexOf(p)
+        if (i == -1) {
+          false
+        } else {
+          buff.remove(i)
+          true
+        }
+      } )
+    } else {
+      false
+    }
   }
 
   def positionAgnosticHashCode: Int = this.positionAgnosticHash

diff --git a/...catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CanonicalizeSuite.scala b/...catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CanonicalizeSuite.scala
@@ -585,4 +585,29 @@ class CanonicalizeSuite extends SparkFunSuite {
     // but canonicalized form should be same even though respective ordering is different
     assert(u1.canonicalized == u2.canonicalized)
   }
+
+  test("SPARK-49618: equality check for repeated legs") {
+    val table1 = LocalRelation(
+      AttributeReference("i", IntegerType)(),
+      AttributeReference("u", DecimalType.SYSTEM_DEFAULT)(),
+      AttributeReference("b", ByteType)(),
+      AttributeReference("d", DoubleType)()).select($"u", $"i").analyze
+
+    val table2 = LocalRelation(
+      AttributeReference("i", IntegerType)(),
+      AttributeReference("u", DecimalType.SYSTEM_DEFAULT)(),
+      AttributeReference("b", ByteType)(),
+      AttributeReference("d", DoubleType)()).select($"u" as "a", $"i" as "b").analyze
+
+    val table3 = LocalRelation(
+      AttributeReference("u", DecimalType.SYSTEM_DEFAULT)(),
+      AttributeReference("d", DoubleType)(),
+      AttributeReference("i", IntegerType)()).select($"u", $"i").analyze
+
+    // These unions are not equal because the head output attributes name are differing
+    val u1 = Union(Seq(table1, table2, table2, table3))
+    val u2 = Union(Seq(table1, table2, table3, table3))
+    assert(u1 != u2)
+    assert(u1.canonicalized != u2.canonicalized)
+  }
 }