fix failed test cases

apache · Jan 23, 2016 · 6a7979d · 6a7979d
1 parent d59b37b
commit 6a7979d
Show file tree

Hide file tree

Showing 3 changed files with 8 additions and 11 deletions.
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
@@ -453,10 +453,8 @@ class Analyzer(
           i.copy(right = dedupRight(left, right))
         case e @ Except(left, right) =>
           e.copy(right = dedupRight(left, right))
-        case u @ Union(left, right) =>
-          u.copy(right = dedupRight(left, right))
-        case cg @ CoGroup(_, _, _, _, _, _, _, _, left, right) =>
-          cg.copy(right = dedupRight(left, right))
+        case cg: CoGroup =>
+          cg.copy(right = dedupRight(cg.left, cg.right))
         case other => other
       }
 

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicOperators.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicOperators.scala
@@ -145,10 +145,6 @@ case class Union(children: Seq[LogicalPlan]) extends LogicalPlan {
     children.length > 1 && childrenResolved && allChildrenCompatible
   }
 
-  def duplicateResolved: Boolean =
-    children.tail.forall( child =>
-      child.outputSet.intersect(children.head.outputSet).isEmpty)
-
   override def statistics: Statistics = {
     val sizeInBytes = children.map(_.statistics.sizeInBytes).sum
     Statistics(sizeInBytes = sizeInBytes)

diff --git a/...atalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/HiveTypeCoercionSuite.scala b/...atalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/HiveTypeCoercionSuite.scala
@@ -421,13 +421,16 @@ class HiveTypeCoercionSuite extends PlanTest {
     assert(r2.left.isInstanceOf[Project])
     assert(r2.right.isInstanceOf[Project])
 
+    // Even if we are doing self Except, we still add Project. The node Except will not be marked
+    // as analyzed unless their exprId are de-duplicated. Thus, the func resolveOperators called in
+    // WidenSetOperationTypes does not skip and return the node before applying the rule.
     val r3 = wt(Except(firstTable, firstTable)).asInstanceOf[Except]
     checkOutput(r3.left, Seq(IntegerType, DecimalType.SYSTEM_DEFAULT, ByteType, DoubleType))
     checkOutput(r3.right, Seq(IntegerType, DecimalType.SYSTEM_DEFAULT, ByteType, DoubleType))
 
-    // Check if no Project is added
-    assert(r3.left.isInstanceOf[LocalRelation])
-    assert(r3.right.isInstanceOf[LocalRelation])
+    // Check if a Project is added
+    assert(r3.left.isInstanceOf[Project])
+    assert(r3.right.isInstanceOf[Project])
   }
 
   test("WidenSetOperationTypes for union") {