address comments

Author: mgaido91

sql/core/src/test/scala/org/apache/spark/sql/execution/PlannerSuite.scala

@@ -704,6 +704,23 @@ class PlannerSuite extends SharedSQLContext {
     df.queryExecution.executedPlan.execute()
   }
 
+  test("SPARK-25278: physical nodes should be different instances for same logical nodes") {
+    val range = Range(1, 1, 1, 1)
+    val df = Union(range, range)
+    val ranges = df.queryExecution.optimizedPlan.collect {
+      case r: Range => r
+    }
+    assert(ranges.length == 2)
+    // Ensure the two Range instances are equal according to their equal method
+    assert(ranges.head == ranges.last)
+    val execRanges = df.queryExecution.sparkPlan.collect {
+      case r: RangeExec => r
+    }
+    assert(execRanges.length == 2)
+    // Ensure the two RangeExec instances are different instances
+    assert(!execRanges.head.eq(execRanges.last))
+  }
+
   test("SPARK-24556: always rewrite output partitioning in ReusedExchangeExec " +
     "and InMemoryTableScanExec") {
     def checkOutputPartitioningRewrite(

sql/core/src/test/scala/org/apache/spark/sql/execution/metric/SQLMetricsSuite.scala

@@ -499,13 +499,15 @@ class SQLMetricsSuite extends SparkFunSuite with SQLMetricsTestUtils with Shared
 
   test("SPARK-25278: output metrics are wrong for plans repeated in the query") {
     val name = "demo_view"
-    sql(s"CREATE OR REPLACE VIEW $name AS VALUES 1,2")
-    val view = spark.table(name)
-    val union = view.union(view)
-    testSparkPlanMetrics(union, 1, Map(
-      0L -> ("Union" -> Map()),
-      1L -> ("LocalTableScan" -> Map("number of output rows" -> 2L)),
-      2L -> ("LocalTableScan" -> Map("number of output rows" -> 2L))))
+    withView(name) {
+      sql(s"CREATE OR REPLACE VIEW $name AS VALUES 1,2")
+      val view = spark.table(name)
+      val union = view.union(view)
+      testSparkPlanMetrics(union, 1, Map(
+        0L -> ("Union" -> Map()),
+        1L -> ("LocalTableScan" -> Map("number of output rows" -> 2L)),
+        2L -> ("LocalTableScan" -> Map("number of output rows" -> 2L))))
+    }
   }
 
   test("writing data out metrics: parquet") {