[SPARK-39319][CORE][SQL] Make query contexts as a part of SparkThrowable

core/src/main/java/org/apache/spark/QueryContext.java

@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark;
+
+import org.apache.spark.annotation.Evolving;
+
+/**
+ * Query context of a {@link SparkThrowable}. It helps users understand where error occur
+ * while executing queries.
+ *
+ * @since 3.4.0
+ */
+@Evolving
+public interface QueryContext {
+    // The object type of the query which throws the exception.
+    // If the exception is directly from the main query, it should be an empty string.
+    // Otherwise, it should be the exact object type in upper case. For example, a "VIEW".
+    String objectType();
+
+    // The object name of the query which throws the exception.
+    // If the exception is directly from the main query, it should be an empty string.
+    // Otherwise, it should be the object name. For example, a view name "V1".
+    String objectName();
+
+    // The starting index in the query text which throws the exception. The index starts from 0.
+    int startIndex();
+
+    // The stopping index in the query which throws the exception. The index starts from 0.
+    int stopIndex();
+
+    // The corresponding fragment of the query which throws the exception.
+    String fragment();
+}

core/src/main/java/org/apache/spark/QueryContextSummary.java

@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark;
+
+import org.apache.spark.annotation.Evolving;
+
+/**
+ * Build textual summary from a query context.
+ *
+ * @since 3.4.0
+ */
+@Evolving
+public interface QueryContextSummary extends QueryContext {
+    // Textual summary of the query context
+    String summary();
+}

core/src/main/java/org/apache/spark/SparkThrowable.java

@@ -59,4 +59,6 @@ default String[] getMessageParameters() {
   default String[] getParameterNames() {
     return SparkThrowableHelper.getParameterNames(this.getErrorClass(), this.getErrorSubClass());
   }
+
+  default QueryContext[] getQueryContext() { return new QueryContext[0]; }
 }

core/src/main/java/org/apache/spark/memory/SparkOutOfMemoryError.java

@@ -39,8 +39,7 @@ public SparkOutOfMemoryError(OutOfMemoryError e) {
     }
 
     public SparkOutOfMemoryError(String errorClass, String[] messageParameters) {
-        super(SparkThrowableHelper.getMessage(errorClass, null,
-                messageParameters, ""));
+        super(SparkThrowableHelper.getMessage(errorClass, null, messageParameters));
         this.errorClass = errorClass;
         this.messageParameters = messageParameters;
     }

core/src/main/scala/org/apache/spark/ErrorInfo.scala

@@ -71,11 +71,18 @@ private[spark] object SparkThrowableHelper {
     mapper.readValue(errorClassesUrl, new TypeReference[SortedMap[String, ErrorInfo]]() {})
   }
 
+  def getMessage(
+      errorClass: String,
+      errorSubClass: String,
+      messageParameters: Array[String]): String = {
+    getMessage(errorClass, errorSubClass, messageParameters, None)
+  }
+
   def getMessage(
       errorClass: String,
       errorSubClass: String,
       messageParameters: Array[String],
-      queryContext: String = ""): String = {
+      context: Option[QueryContextSummary]): String = {
     val errorInfo = errorClassToInfoMap.getOrElse(errorClass,
       throw new IllegalArgumentException(s"Cannot find error class '$errorClass'"))
     val (displayClass, displayMessageParameters, displayFormat) = if (errorInfo.subClass.isEmpty) {
@@ -93,11 +100,8 @@ private[spark] object SparkThrowableHelper {
     val displayMessage = String.format(
       displayFormat.replaceAll("<[a-zA-Z0-9_-]+>", "%s"),
       displayMessageParameters : _*)
-    val displayQueryContext = if (queryContext.isEmpty) {
-      ""
-    } else {
-      s"\n$queryContext"
-    }
+    val displayQueryContext = context.map(q => "\n" + q.summary()).getOrElse("")
+
     s"[$displayClass] $displayMessage$displayQueryContext"
   }
 

core/src/main/scala/org/apache/spark/SparkException.scala

@@ -119,15 +119,16 @@ private[spark] class SparkArithmeticException(
     errorClass: String,
     errorSubClass: Option[String] = None,
     messageParameters: Array[String],
-    queryContext: String = "")
+    context: Option[QueryContextSummary] = None)
   extends ArithmeticException(
-    SparkThrowableHelper.getMessage(errorClass, errorSubClass.orNull,
-      messageParameters, queryContext))
+    SparkThrowableHelper.getMessage(errorClass, errorSubClass.orNull, messageParameters, context))
     with SparkThrowable {
 
   override def getMessageParameters: Array[String] = messageParameters
   override def getErrorClass: String = errorClass
-  override def getErrorSubClass: String = errorSubClass.orNull}
+  override def getErrorSubClass: String = errorSubClass.orNull
+  override def getQueryContext: Array[QueryContext] = context.toArray
+}
 
 /**
  * Unsupported operation exception thrown from Spark with an error class.
@@ -193,15 +194,16 @@ private[spark] class SparkDateTimeException(
     errorClass: String,
     errorSubClass: Option[String] = None,
     messageParameters: Array[String],
-    queryContext: String = "")
+    context: Option[QueryContextSummary] = None)
   extends DateTimeException(
-    SparkThrowableHelper.getMessage(errorClass, errorSubClass.orNull,
-      messageParameters, queryContext))
+    SparkThrowableHelper.getMessage(errorClass, errorSubClass.orNull, messageParameters, context))
     with SparkThrowable {
 
   override def getMessageParameters: Array[String] = messageParameters
   override def getErrorClass: String = errorClass
-  override def getErrorSubClass: String = errorSubClass.orNull}
+  override def getErrorSubClass: String = errorSubClass.orNull
+  override def getQueryContext: Array[QueryContext] = context.toArray
+}
 
 /**
  * Hadoop file already exists exception thrown from Spark with an error class.
@@ -240,15 +242,16 @@ private[spark] class SparkNumberFormatException(
     errorClass: String,
     errorSubClass: Option[String] = None,
     messageParameters: Array[String],
-    queryContext: String)
+    context: Option[QueryContextSummary])
   extends NumberFormatException(
-    SparkThrowableHelper.getMessage(errorClass, errorSubClass.orNull,
-      messageParameters, queryContext))
+    SparkThrowableHelper.getMessage(errorClass, errorSubClass.orNull, messageParameters, context))
     with SparkThrowable {
 
   override def getMessageParameters: Array[String] = messageParameters
   override def getErrorClass: String = errorClass
-  override def getErrorSubClass: String = errorSubClass.orNull}
+  override def getErrorSubClass: String = errorSubClass.orNull
+  override def getQueryContext: Array[QueryContext] = context.toArray
+}
 
 /**
  * No such method exception thrown from Spark with an error class.
@@ -317,23 +320,22 @@ private[spark] class SparkRuntimeException(
     errorSubClass: Option[String] = None,
     messageParameters: Array[String],
     cause: Throwable = null,
-    queryContext: String = "")
+    context: Option[QueryContextSummary] = None)
   extends RuntimeException(
-    SparkThrowableHelper.getMessage(errorClass, errorSubClass.orNull,
-      messageParameters, queryContext),
+    SparkThrowableHelper.getMessage(errorClass, errorSubClass.orNull, messageParameters, context),
     cause)
     with SparkThrowable {
 
   def this(errorClass: String,
            errorSubClass: String,
            messageParameters: Array[String],
            cause: Throwable,
-           queryContext: String)
+           context: Option[QueryContextSummary])
     = this(errorClass = errorClass,
     errorSubClass = Some(errorSubClass),
     messageParameters = messageParameters,
     cause = cause,
-    queryContext = queryContext)
+    context = context)
 
   def this(errorClass: String,
            errorSubClass: String,
@@ -342,11 +344,12 @@ private[spark] class SparkRuntimeException(
     errorSubClass = Some(errorSubClass),
     messageParameters = messageParameters,
     cause = null,
-    queryContext = "")
+    context = None)
 
   override def getMessageParameters: Array[String] = messageParameters
   override def getErrorClass: String = errorClass
   override def getErrorSubClass: String = errorSubClass.orNull
+  override def getQueryContext: Array[QueryContext] = context.toArray
 }
 
 /**
@@ -372,16 +375,15 @@ private[spark] class SparkArrayIndexOutOfBoundsException(
     errorClass: String,
     errorSubClass: Option[String] = None,
     messageParameters: Array[String],
-    queryContext: String)
+    context: Option[QueryContextSummary])
   extends ArrayIndexOutOfBoundsException(
-    // scalastyle:off line.size.limit
-    SparkThrowableHelper.getMessage(errorClass, errorSubClass.orNull, messageParameters, queryContext))
-    // scalastyle:on line.size.limit
+    SparkThrowableHelper.getMessage(errorClass, errorSubClass.orNull, messageParameters, context))
     with SparkThrowable {
 
   override def getMessageParameters: Array[String] = messageParameters
   override def getErrorClass: String = errorClass
   override def getErrorSubClass: String = errorSubClass.orNull
+  override def getQueryContext: Array[QueryContext] = context.toArray
 }
 
 /**
@@ -413,15 +415,15 @@ private[spark] class SparkNoSuchElementException(
     errorClass: String,
     errorSubClass: Option[String] = None,
     messageParameters: Array[String],
-    queryContext: String)
+    context: Option[QueryContextSummary])
   extends NoSuchElementException(
-    SparkThrowableHelper.getMessage(errorClass, errorSubClass.orNull,
-      messageParameters, queryContext))
+    SparkThrowableHelper.getMessage(errorClass, errorSubClass.orNull, messageParameters, context))
     with SparkThrowable {
 
   override def getMessageParameters: Array[String] = messageParameters
   override def getErrorClass: String = errorClass
   override def getErrorSubClass: String = errorSubClass.orNull
+  override def getQueryContext: Array[QueryContext] = context.toArray
 }
 
 /**

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

@@ -25,7 +25,7 @@ import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.analysis.{TypeCheckResult, TypeCoercion}
 import org.apache.spark.sql.catalyst.expressions.codegen._
 import org.apache.spark.sql.catalyst.expressions.codegen.Block._
-import org.apache.spark.sql.catalyst.trees.TreeNodeTag
+import org.apache.spark.sql.catalyst.trees.{SqlQueryContext, TreeNodeTag}
 import org.apache.spark.sql.catalyst.trees.TreePattern._
 import org.apache.spark.sql.catalyst.util._
 import org.apache.spark.sql.catalyst.util.DateTimeConstants._
@@ -481,10 +481,10 @@ case class Cast(
 
   override def nullable: Boolean = child.nullable || Cast.forceNullable(child.dataType, dataType)
 
-  override def initQueryContext(): String = if (ansiEnabled) {
-    origin.context
+  override def initQueryContext(): Option[SqlQueryContext] = if (ansiEnabled) {
+    Some(origin.context)
   } else {
-    ""
+    None
   }
 
   // When this cast involves TimeZone, it's only resolved if the timeZoneId is set;
@@ -995,9 +995,12 @@ case class Cast(
    * If overflow occurs, if `spark.sql.ansi.enabled` is false, null is returned;
    * otherwise, an `ArithmeticException` is thrown.
    */
-  private[this] def toPrecision(value: Decimal, decimalType: DecimalType): Decimal =
+  private[this] def toPrecision(
+      value: Decimal,
+      decimalType: DecimalType,
+      context: Option[SqlQueryContext]): Decimal =
     value.toPrecision(
-      decimalType.precision, decimalType.scale, Decimal.ROUND_HALF_UP, !ansiEnabled)
+      decimalType.precision, decimalType.scale, Decimal.ROUND_HALF_UP, !ansiEnabled, context)
 
 
   private[this] def castToDecimal(from: DataType, target: DecimalType): Any => Any = from match {
@@ -1010,14 +1013,15 @@ case class Cast(
       buildCast[UTF8String](_,
         s => changePrecision(Decimal.fromStringANSI(s, target, queryContext), target))
     case BooleanType =>
-      buildCast[Boolean](_, b => toPrecision(if (b) Decimal.ONE else Decimal.ZERO, target))
+      buildCast[Boolean](_,
+        b => toPrecision(if (b) Decimal.ONE else Decimal.ZERO, target, queryContext))
     case DateType =>
       buildCast[Int](_, d => null) // date can't cast to decimal in Hive
     case TimestampType =>
       // Note that we lose precision here.
       buildCast[Long](_, t => changePrecision(Decimal(timestampToDouble(t)), target))
     case dt: DecimalType =>
-      b => toPrecision(b.asInstanceOf[Decimal], target)
+      b => toPrecision(b.asInstanceOf[Decimal], target, queryContext)
     case t: IntegralType =>
       b => changePrecision(Decimal(t.integral.asInstanceOf[Integral[Any]].toLong(b)), target)
     case x: FractionalType =>

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

@@ -24,7 +24,7 @@ import org.apache.spark.sql.catalyst.analysis.{FunctionRegistry, TypeCheckResult
 import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateFunction
 import org.apache.spark.sql.catalyst.expressions.codegen._
 import org.apache.spark.sql.catalyst.expressions.codegen.Block._
-import org.apache.spark.sql.catalyst.trees.{BinaryLike, LeafLike, QuaternaryLike, TernaryLike, TreeNode, UnaryLike}
+import org.apache.spark.sql.catalyst.trees.{BinaryLike, LeafLike, QuaternaryLike, SqlQueryContext, TernaryLike, TreeNode, UnaryLike}
 import org.apache.spark.sql.catalyst.trees.TreePattern.{RUNTIME_REPLACEABLE, TreePattern}
 import org.apache.spark.sql.catalyst.util.truncatedString
 import org.apache.spark.sql.errors.QueryExecutionErrors
@@ -593,9 +593,9 @@ abstract class UnaryExpression extends Expression with UnaryLike[Expression] {
  * to executors. It will also be kept after rule transforms.
  */
 trait SupportQueryContext extends Expression with Serializable {
-  protected var queryContext: String = initQueryContext()
+  protected var queryContext: Option[SqlQueryContext] = initQueryContext()
 
-  def initQueryContext(): String
+  def initQueryContext(): Option[SqlQueryContext]
 
   // Note: Even though query contexts are serialized to executors, it will be regenerated from an
   //       empty "Origin" during rule transforms since "Origin"s are not serialized to executors