[SPARK-51388][SQL] Improve SQL fragment propagation in to_timestamp and UNION

### What changes were proposed in this pull request?

Improve SQL fragment propagation in to_timestamp and UNION.

### Why are the changes needed?

To improve error messages:
![image](https://github.com/user-attachments/assets/4952ce49-5a29-4bf7-9d9f-0e443b1127ec)

### Does this PR introduce _any_ user-facing change?

Invalid `Cast` error messages become better.

### How was this patch tested?

Golden file tests are regenerated.

### Was this patch authored or co-authored using generative AI tooling?

No.

Closes #50154 from vladimirg-db/vladimir-golubev_data/improve-sql-fragment-in-some-places.

Authored-by: Vladimir Golubev <vladimir.golubev@databricks.com>
Signed-off-by: Wenchen Fan <wenchen@databricks.com>

Author: vladimirg-db

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TypeCoercionBase.scala

@@ -41,6 +41,7 @@ import org.apache.spark.sql.catalyst.plans.logical.{
   Unpivot
 }
 import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.trees.CurrentOrigin.withOrigin
 import org.apache.spark.sql.connector.catalog.procedures.BoundProcedure
 import org.apache.spark.sql.types.DataType
 
@@ -142,7 +143,9 @@ abstract class TypeCoercionBase extends TypeCoercionHelper {
         case s @ Except(left, right, isAll)
             if s.childrenResolved &&
             left.output.length == right.output.length && !s.resolved =>
-          val newChildren: Seq[LogicalPlan] = buildNewChildrenWithWiderTypes(left :: right :: Nil)
+          val newChildren: Seq[LogicalPlan] = withOrigin(s.origin) {
+            buildNewChildrenWithWiderTypes(left :: right :: Nil)
+          }
           if (newChildren.isEmpty) {
             s -> Nil
           } else {
@@ -154,7 +157,9 @@ abstract class TypeCoercionBase extends TypeCoercionHelper {
         case s @ Intersect(left, right, isAll)
             if s.childrenResolved &&
             left.output.length == right.output.length && !s.resolved =>
-          val newChildren: Seq[LogicalPlan] = buildNewChildrenWithWiderTypes(left :: right :: Nil)
+          val newChildren: Seq[LogicalPlan] = withOrigin(s.origin) {
+            buildNewChildrenWithWiderTypes(left :: right :: Nil)
+          }
           if (newChildren.isEmpty) {
             s -> Nil
           } else {
@@ -166,7 +171,9 @@ abstract class TypeCoercionBase extends TypeCoercionHelper {
         case s: Union
             if s.childrenResolved && !s.byName &&
             s.children.forall(_.output.length == s.children.head.output.length) && !s.resolved =>
-          val newChildren: Seq[LogicalPlan] = buildNewChildrenWithWiderTypes(s.children)
+          val newChildren: Seq[LogicalPlan] = withOrigin(s.origin) {
+            buildNewChildrenWithWiderTypes(s.children)
+          }
           if (newChildren.isEmpty) {
             s -> Nil
           } else {

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/datetimeExpressions.scala

@@ -29,6 +29,7 @@ import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.analysis.{ExpressionBuilder, FunctionRegistry}
 import org.apache.spark.sql.catalyst.expressions.codegen._
 import org.apache.spark.sql.catalyst.expressions.codegen.Block._
+import org.apache.spark.sql.catalyst.trees.CurrentOrigin.withOrigin
 import org.apache.spark.sql.catalyst.trees.TreePattern._
 import org.apache.spark.sql.catalyst.util.{DateTimeUtils, LegacyDateFormats, TimestampFormatter}
 import org.apache.spark.sql.catalyst.util.DateTimeConstants._
@@ -2104,11 +2105,13 @@ case class ParseToDate(
     ansiEnabled: Boolean = SQLConf.get.ansiEnabled)
   extends RuntimeReplaceable with ImplicitCastInputTypes with TimeZoneAwareExpression {
 
-  override lazy val replacement: Expression = format.map { f =>
-    Cast(GetTimestamp(left, f, TimestampType, "try_to_date", timeZoneId, ansiEnabled), DateType,
-      timeZoneId, EvalMode.fromBoolean(ansiEnabled))
-  }.getOrElse(Cast(left, DateType, timeZoneId,
-    EvalMode.fromBoolean(ansiEnabled))) // backwards compatibility
+  override lazy val replacement: Expression = withOrigin(origin) {
+    format.map { f =>
+      Cast(GetTimestamp(left, f, TimestampType, "try_to_date", timeZoneId, ansiEnabled), DateType,
+        timeZoneId, EvalMode.fromBoolean(ansiEnabled))
+    }.getOrElse(Cast(left, DateType, timeZoneId,
+      EvalMode.fromBoolean(ansiEnabled))) // backwards compatibility
+  }
 
   def this(left: Expression, format: Expression) = {
     this(left, Option(format))
@@ -2183,9 +2186,11 @@ case class ParseToTimestamp(
     failOnError: Boolean = SQLConf.get.ansiEnabled)
   extends RuntimeReplaceable with ImplicitCastInputTypes with TimeZoneAwareExpression {
 
-  override lazy val replacement: Expression = format.map { f =>
-    GetTimestamp(left, f, dataType, "try_to_timestamp", timeZoneId, failOnError = failOnError)
-  }.getOrElse(Cast(left, dataType, timeZoneId, ansiEnabled = failOnError))
+  override lazy val replacement: Expression = withOrigin(origin) {
+    format.map { f =>
+      GetTimestamp(left, f, dataType, "try_to_timestamp", timeZoneId, failOnError = failOnError)
+    }.getOrElse(Cast(left, dataType, timeZoneId, ansiEnabled = failOnError))
+  }
 
   def this(left: Expression, format: Expression) = {
     this(left, Option(format), SQLConf.get.timestampType)

sql/core/src/test/resources/sql-tests/results/timestampNTZ/timestamp-ansi.sql.out

@@ -394,7 +394,9 @@ org.apache.spark.SparkDateTimeException
   "queryContext" : [ {
     "objectType" : "",
     "objectName" : "",
-    "fragment" : ""
+    "startIndex" : 8,
+    "stopIndex" : 22,
+    "fragment" : "to_timestamp(1)"
   } ]
 }
 

sql/core/src/test/resources/sql-tests/results/typeCoercion/native/stringCastAndExpressions.sql.out

@@ -344,7 +344,9 @@ org.apache.spark.SparkDateTimeException
   "queryContext" : [ {
     "objectType" : "",
     "objectName" : "",
-    "fragment" : ""
+    "startIndex" : 8,
+    "stopIndex" : 22,
+    "fragment" : "to_timestamp(a)"
   } ]
 }
 

sql/core/src/test/resources/sql-tests/results/udf/udf-union.sql.out

@@ -51,9 +51,9 @@ org.apache.spark.SparkNumberFormatException
   "queryContext" : [ {
     "objectType" : "",
     "objectName" : "",
-    "startIndex" : 1,
-    "stopIndex" : 243,
-    "fragment" : "SELECT udf(c1) as c1, udf(c2) as c2\nFROM   (SELECT udf(c1) as c1, udf(c2) as c2 FROM t1 WHERE c2 = 'a'\n        UNION ALL\n        SELECT udf(c1) as c1, udf(c2) as c2 FROM t2\n        UNION ALL\n        SELECT udf(c1) as c1, udf(c2) as c2 FROM t2)"
+    "startIndex" : 45,
+    "stopIndex" : 172,
+    "fragment" : "SELECT udf(c1) as c1, udf(c2) as c2 FROM t1 WHERE c2 = 'a'\n        UNION ALL\n        SELECT udf(c1) as c1, udf(c2) as c2 FROM t2"
   } ]
 }
 

sql/core/src/test/resources/sql-tests/results/union.sql.out

@@ -51,9 +51,9 @@ org.apache.spark.SparkNumberFormatException
   "queryContext" : [ {
     "objectType" : "",
     "objectName" : "",
-    "startIndex" : 1,
-    "stopIndex" : 133,
-    "fragment" : "SELECT *\nFROM   (SELECT * FROM t1 where c1 = 1\n        UNION ALL\n        SELECT * FROM t2\n        UNION ALL\n        SELECT * FROM t2)"
+    "startIndex" : 18,
+    "stopIndex" : 89,
+    "fragment" : "SELECT * FROM t1 where c1 = 1\n        UNION ALL\n        SELECT * FROM t2"
   } ]
 }