fix

Author: mihailotim-db

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala

@@ -614,13 +614,20 @@ case class Union(
     Some(sum.toLong)
   }
 
+  def duplicateResolvedPerChild: Boolean =
+    children.forall(child => child.outputSet.size == child.output.size)
+
   def duplicateResolved: Boolean = {
     children.map(_.outputSet.size).sum ==
       AttributeSet.fromAttributeSets(children.map(_.outputSet)).size
   }
 
   override lazy val resolved: Boolean = {
-    children.length > 1 && !(byName || allowMissingCol) && childrenResolved && allChildrenCompatible
+    children.length > 1 &&
+    !(byName || allowMissingCol) &&
+    childrenResolved &&
+    allChildrenCompatible &&
+    duplicateResolvedPerChild
   }
 
   override protected def withNewChildrenInternal(newChildren: IndexedSeq[LogicalPlan]): Union =

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

@@ -4963,6 +4963,23 @@ class SQLQuerySuite extends QueryTest with SharedSparkSession with AdaptiveSpark
     )
   }
 
+  test("SPARK-52686: Union should be resolved only if there are no duplicates") {
+    withTable("t1") {
+      withView("v1") {
+        sql("CREATE TABLE t1(col1 STRING)")
+        sql("CREATE VIEW v1 AS SELECT * FROM t1")
+        val analyzedPlan = sql(
+          "SELECT * FROM (SELECT col1, col1 FROM v1 UNION SELECT col1, col1 FROM v1)"
+        ).queryExecution.analyzed
+
+        // Resolving * should wait for Union output to be deduplicated, otherwise we are
+        // left with duplicate ExprIds in the result.
+        val exprIds = analyzedPlan.asInstanceOf[Project].projectList.map(_.exprId)
+        assert(exprIds.size == exprIds.distinct.size)
+      }
+    }
+  }
+
   Seq(true, false).foreach { codegenEnabled =>
     test(s"SPARK-52060: one row relation with codegen enabled - $codegenEnabled") {
       withSQLConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> codegenEnabled.toString) {