[SPARK-14142][SQL] Replace internal use of unionAll with union

## What changes were proposed in this pull request?
unionAll has been deprecated in SPARK-14088.

## How was this patch tested?
Should be covered by all existing tests.

Author: Reynold Xin <[email protected]>

Closes apache#11946 from rxin/SPARK-14142.

Author: rxin

python/pyspark/sql/dataframe.py

@@ -360,7 +360,7 @@ def repartition(self, numPartitions, *cols):
 
         >>> df.repartition(10).rdd.getNumPartitions()
         10
-        >>> data = df.unionAll(df).repartition("age")
+        >>> data = df.union(df).repartition("age")
         >>> data.show()
         +---+-----+
         |age| name|
@@ -919,7 +919,7 @@ def union(self, other):
         This is equivalent to `UNION ALL` in SQL. To do a SQL-style set union
         (that does deduplication of elements), use this function followed by a distinct.
         """
-        return DataFrame(self._jdf.unionAll(other._jdf), self.sql_ctx)
+        return DataFrame(self._jdf.union(other._jdf), self.sql_ctx)
 
     @since(1.3)
     def unionAll(self, other):

python/pyspark/sql/tests.py

@@ -599,7 +599,7 @@ def test_parquet_with_udt(self):
         point = df1.head().point
         self.assertEqual(point, PythonOnlyPoint(1.0, 2.0))
 
-    def test_unionAll_with_udt(self):
+    def test_union_with_udt(self):
         from pyspark.sql.tests import ExamplePoint, ExamplePointUDT
         row1 = (1.0, ExamplePoint(1.0, 2.0))
         row2 = (2.0, ExamplePoint(3.0, 4.0))
@@ -608,7 +608,7 @@ def test_unionAll_with_udt(self):
         df1 = self.sqlCtx.createDataFrame([row1], schema)
         df2 = self.sqlCtx.createDataFrame([row2], schema)
 
-        result = df1.unionAll(df2).orderBy("label").collect()
+        result = df1.union(df2).orderBy("label").collect()
         self.assertEqual(
             result,
             [

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/dsl/package.scala

@@ -280,7 +280,7 @@ package object dsl {
 
       def intersect(otherPlan: LogicalPlan): LogicalPlan = Intersect(logicalPlan, otherPlan)
 
-      def unionAll(otherPlan: LogicalPlan): LogicalPlan = Union(logicalPlan, otherPlan)
+      def union(otherPlan: LogicalPlan): LogicalPlan = Union(logicalPlan, otherPlan)
 
       def generate(
         generator: Generator,

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisErrorSuite.scala

@@ -250,7 +250,7 @@ class AnalysisErrorSuite extends AnalysisTest {
 
   errorTest(
     "union with unequal number of columns",
-    testRelation.unionAll(testRelation2),
+    testRelation.union(testRelation2),
     "union" :: "number of columns" :: testRelation2.output.length.toString ::
       testRelation.output.length.toString :: Nil)
 

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisSuite.scala

@@ -32,7 +32,7 @@ class AnalysisSuite extends AnalysisTest {
     val plan = (1 to 100)
       .map(_ => testRelation)
       .fold[LogicalPlan](testRelation) { (a, b) =>
-        a.select(UnresolvedStar(None)).select('a).unionAll(b.select(UnresolvedStar(None)))
+        a.select(UnresolvedStar(None)).select('a).union(b.select(UnresolvedStar(None)))
       }
 
     assertAnalysisSuccess(plan)

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/PruneFiltersSuite.scala

@@ -60,8 +60,8 @@ class PruneFiltersSuite extends PlanTest {
 
     val query =
       tr1.where('a.attr > 10)
-        .unionAll(tr2.where('d.attr > 10)
-        .unionAll(tr3.where('g.attr > 10)))
+        .union(tr2.where('d.attr > 10)
+        .union(tr3.where('g.attr > 10)))
     val queryWithUselessFilter = query.where('a.attr > 10)
 
     val optimized = Optimize.execute(queryWithUselessFilter.analyze)

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/ConstraintPropagationSuite.scala

@@ -109,14 +109,14 @@ class ConstraintPropagationSuite extends SparkFunSuite {
 
     assert(tr1
       .where('a.attr > 10)
-      .unionAll(tr2.where('e.attr > 10)
-      .unionAll(tr3.where('i.attr > 10)))
+      .union(tr2.where('e.attr > 10)
+      .union(tr3.where('i.attr > 10)))
       .analyze.constraints.isEmpty)
 
     verifyConstraints(tr1
       .where('a.attr > 10)
-      .unionAll(tr2.where('d.attr > 10)
-      .unionAll(tr3.where('g.attr > 10)))
+      .union(tr2.where('d.attr > 10)
+      .union(tr3.where('g.attr > 10)))
       .analyze.constraints,
       ExpressionSet(Seq(resolveColumn(tr1, "a") > 10,
         IsNotNull(resolveColumn(tr1, "a")))))

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/memory.scala

@@ -97,7 +97,7 @@ case class MemoryStream[A : Encoder](id: Int, sqlContext: SQLContext)
       s"MemoryBatch [$startOrdinal, $endOrdinal]: ${newBlocks.flatMap(_.collect()).mkString(", ")}")
     newBlocks
       .map(_.toDF())
-      .reduceOption(_ unionAll _)
+      .reduceOption(_ union _)
       .getOrElse {
         sys.error("No data selected!")
       }

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

@@ -363,7 +363,7 @@ class CachedTableSuite extends QueryTest with SQLTestUtils with SharedSQLContext
   }
 
   test("A cached table preserves the partitioning and ordering of its cached SparkPlan") {
-    val table3x = testData.unionAll(testData).unionAll(testData)
+    val table3x = testData.union(testData).union(testData)
     table3x.registerTempTable("testData3x")
 
     sql("SELECT key, value FROM testData3x ORDER BY key").registerTempTable("orderedTable")

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

@@ -57,7 +57,7 @@ class DataFrameStatSuite extends QueryTest with SharedSQLContext {
       val splits = data.randomSplit(Array[Double](1, 2, 3), seed)
       assert(splits.length == 3, "wrong number of splits")
 
-      assert(splits.reduce((a, b) => a.unionAll(b)).sort("id").collect().toList ==
+      assert(splits.reduce((a, b) => a.union(b)).sort("id").collect().toList ==
         data.collect().toList, "incomplete or wrong split")
 
       val s = splits.map(_.count())