[SPARK-44507][SQL][CONNECT] Move AnalysisException to sql/api

docs/sql-migration-guide.md

@@ -28,7 +28,7 @@ license: |
 - Since Spark 3.5, Spark thrift server will interrupt task when canceling a running statement. To restore the previous behavior, set `spark.sql.thriftServer.interruptOnCancel` to `false`.
 - Since Spark 3.5, the Avro will throw `AnalysisException` when reading Interval types as Date or Timestamp types, or reading Decimal types with lower precision. To restore the legacy behavior, set `spark.sql.legacy.avro.allowIncompatibleSchema` to `true`
 - Since Spark 3.5, Row's json and prettyJson methods are moved to `ToJsonUtil`.
-- Since Spark 3.5, ParseException is a subclass of SparkException instead of AnalysisException.
+- Since Spark 3.5, the `plan` field is moved from `AnalysisException` to `EnhancedAnalysisException`.
 
 ## Upgrading from Spark SQL 3.3 to 3.4
 

project/MimaExcludes.scala

@@ -65,12 +65,15 @@ object MimaExcludes {
     ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.status.api.v1.JobData.this"),
     // [SPARK-44205][SQL] Extract Catalyst Code from DecimalType
     ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.sql.types.DecimalType.unapply"),
+    // [SPARK-44507][SQL][CONNECT] Move AnalysisException to sql/api.
+    ProblemFilters.exclude[MissingClassProblem]("org.apache.spark.sql.AnalysisException"),
+    ProblemFilters.exclude[MissingClassProblem]("org.apache.spark.sql.AnalysisException$"),
     // [SPARK-44535][CONNECT][SQL] Move required Streaming API to sql/api
     ProblemFilters.exclude[MissingClassProblem]("org.apache.spark.sql.streaming.GroupStateTimeout"),
     ProblemFilters.exclude[MissingClassProblem]("org.apache.spark.sql.streaming.OutputMode")
   )
 
-  // Defulat exclude rules
+  // Default exclude rules
   lazy val defaultExcludes = Seq(
     // Spark Internals
     ProblemFilters.exclude[Problem]("org.apache.spark.rpc.*"),

sql/catalyst/src/main/scala/org/apache/spark/sql/AnalysisException.scala → sql/api/src/main/scala/org/apache/spark/sql/AnalysisException.scala

@@ -21,7 +21,6 @@ import scala.collection.JavaConverters._
 
 import org.apache.spark.{QueryContext, SparkThrowable, SparkThrowableHelper}
 import org.apache.spark.annotation.Stable
-import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
 import org.apache.spark.sql.catalyst.trees.{Origin, WithOrigin}
 
 /**
@@ -34,8 +33,6 @@ class AnalysisException protected[sql] (
     val message: String,
     val line: Option[Int] = None,
     val startPosition: Option[Int] = None,
-    // Some plans fail to serialize due to bugs in scala collections.
-    @transient val plan: Option[LogicalPlan] = None,
     val cause: Option[Throwable] = None,
     val errorClass: Option[String] = None,
     val messageParameters: Map[String, String] = Map.empty,
@@ -102,12 +99,11 @@ class AnalysisException protected[sql] (
       message: String = this.message,
       line: Option[Int] = this.line,
       startPosition: Option[Int] = this.startPosition,
-      plan: Option[LogicalPlan] = this.plan,
       cause: Option[Throwable] = this.cause,
       errorClass: Option[String] = this.errorClass,
       messageParameters: Map[String, String] = this.messageParameters,
       context: Array[QueryContext] = this.context): AnalysisException =
-    new AnalysisException(message, line, startPosition, plan, cause, errorClass,
+    new AnalysisException(message, line, startPosition, cause, errorClass,
       messageParameters, context)
 
   def withPosition(origin: Origin): AnalysisException = {
@@ -119,10 +115,7 @@ class AnalysisException protected[sql] (
     newException
   }
 
-  override def getMessage: String = {
-    val planAnnotation = Option(plan).flatten.map(p => s";\n$p").getOrElse("")
-    getSimpleMessage + planAnnotation
-  }
+  override def getMessage: String = getSimpleMessage
 
   // Outputs an exception without the logical plan.
   // For testing only

sql/api/src/main/scala/org/apache/spark/sql/catalyst/parser/parsers.scala

@@ -23,9 +23,9 @@ import org.antlr.v4.runtime.atn.PredictionMode
 import org.antlr.v4.runtime.misc.{Interval, ParseCancellationException}
 import org.antlr.v4.runtime.tree.TerminalNodeImpl
 
-import org.apache.spark.{QueryContext, SparkException, SparkThrowable, SparkThrowableHelper}
+import org.apache.spark.{QueryContext, SparkThrowable, SparkThrowableHelper}
 import org.apache.spark.internal.Logging
-import org.apache.spark.sql.SqlApiConf
+import org.apache.spark.sql.{AnalysisException, SqlApiConf}
 import org.apache.spark.sql.catalyst.trees.{CurrentOrigin, Origin, WithOrigin}
 import org.apache.spark.sql.catalyst.util.SparkParserUtils
 import org.apache.spark.sql.errors.QueryParsingErrors
@@ -184,15 +184,17 @@ case object ParseErrorListener extends BaseErrorListener {
  */
 class ParseException(
     val command: Option[String],
-    val message: String,
+    message: String,
     val start: Origin,
     val stop: Origin,
-    val errorClass: Option[String] = None,
-    val messageParameters: Map[String, String] = Map.empty,
-    val queryContext: Array[QueryContext] = ParseException.getQueryContext())
-  extends SparkException(
+    errorClass: Option[String] = None,
+    messageParameters: Map[String, String] = Map.empty,
+    queryContext: Array[QueryContext] = ParseException.getQueryContext())
+  extends AnalysisException(
     message,
-    cause = null,
+    start.line,
+    start.startPosition,
+    None,
     errorClass,
     messageParameters,
     queryContext) {
@@ -222,14 +224,6 @@ class ParseException(
       Some(errorClass),
       messageParameters)
 
-  // Methods added to retain compatibility with AnalysisException.
-  @deprecated("Use start.line instead.")
-  def line: Option[Int] = start.line
-  @deprecated("Use start.startPosition instead.")
-  def startPosition: Option[Int] = start.startPosition
-  @deprecated("ParseException is never caused by another exception.")
-  def cause: Option[Throwable] = None
-
   override def getMessage: String = {
     val builder = new StringBuilder
     builder ++= "\n" ++= message

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/ExtendedAnalysisException.scala

@@ -0,0 +1,87 @@
+/*
+ * 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.sql.catalyst
+
+import org.apache.spark.QueryContext
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+
+/**
+ * Internal [[AnalysisException]] that also captures a [[LogicalPlan]].
+ */
+class ExtendedAnalysisException(
+    message: String,
+    line: Option[Int] = None,
+    startPosition: Option[Int] = None,
+    // Some plans fail to serialize due to bugs in scala collections.
+    @transient val plan: Option[LogicalPlan] = None,
+    cause: Option[Throwable] = None,
+    errorClass: Option[String] = None,
+    messageParameters: Map[String, String] = Map.empty,
+    context: Array[QueryContext] = Array.empty)
+  extends AnalysisException(
+    message,
+    line,
+    startPosition,
+    cause,
+    errorClass,
+    messageParameters,
+    context) {
+
+  def this(e: AnalysisException, plan: LogicalPlan) = {
+    this(
+      e.message,
+      e.line,
+      e.startPosition,
+      Option(plan),
+      e.cause,
+      e.errorClass,
+      e.messageParameters,
+      e.context)
+    setStackTrace(e.getStackTrace)
+  }
+
+  override def copy(
+      message: String,
+      line: Option[Int],
+      startPosition: Option[Int],
+      cause: Option[Throwable],
+      errorClass: Option[String],
+      messageParameters: Map[String, String],
+      context: Array[QueryContext]): ExtendedAnalysisException = {
+    new ExtendedAnalysisException(message, line, startPosition, plan, cause, errorClass,
+      messageParameters, context)
+  }
+
+  override def getMessage: String = {
+    val planAnnotation = Option(plan).flatten.map(p => s";\n$p").getOrElse("")
+    getSimpleMessage + planAnnotation
+  }
+
+  override def toString: String = {
+    val message = getLocalizedMessage
+    if (message != null) {
+      ExtendedAnalysisException.name + ": " + message
+    } else {
+      ExtendedAnalysisException.name
+    }
+  }
+}
+
+object ExtendedAnalysisException {
+  private val name = classOf[AnalysisException].getName
+}

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

@@ -29,7 +29,7 @@ import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.catalyst._
 import org.apache.spark.sql.catalyst.catalog._
 import org.apache.spark.sql.catalyst.encoders.OuterScopes
-import org.apache.spark.sql.catalyst.expressions.{Expression, FrameLessOffsetWindowFunction, _}
+import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.expressions.SubExprUtils._
 import org.apache.spark.sql.catalyst.expressions.aggregate._
 import org.apache.spark.sql.catalyst.expressions.objects._
@@ -212,9 +212,7 @@ class Analyzer(override val catalogManager: CatalogManager) extends RuleExecutor
         analyzed
       } catch {
         case e: AnalysisException =>
-          val ae = e.copy(plan = Option(analyzed))
-          ae.setStackTrace(e.getStackTrace)
-          throw ae
+          throw new ExtendedAnalysisException(e, analyzed)
       }
     }
   }

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

@@ -17,7 +17,7 @@
 
 package org.apache.spark.sql.catalyst.analysis
 
-import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.ExtendedAnalysisException
 import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeReference, CaseWhen, Cast, CreateNamedStruct, Expression, GetStructField, IsNotNull, LessThan, Literal, PreciseTimestampConversion, SessionWindow, Subtract, TimeWindow, WindowTime}
 import org.apache.spark.sql.catalyst.plans.logical.{Expand, Filter, LogicalPlan, Project}
 import org.apache.spark.sql.catalyst.rules.Rule
@@ -310,7 +310,7 @@ object ResolveWindowTime extends Rule[LogicalPlan] {
           if (!metadata.contains(TimeWindow.marker) &&
             !metadata.contains(SessionWindow.marker)) {
             // FIXME: error framework?
-            throw new AnalysisException(
+            throw new ExtendedAnalysisException(
               s"The input is not a correct window column: $windowTime", plan = Some(p))
           }
 

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

@@ -19,6 +19,7 @@ package org.apache.spark.sql.catalyst.analysis
 
 import org.apache.spark.internal.Logging
 import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.ExtendedAnalysisException
 import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference, CurrentDate, CurrentTimestampLike, Expression, GroupingSets, LocalTimestamp, MonotonicallyIncreasingID, SessionWindow, WindowExpression}
 import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression
 import org.apache.spark.sql.catalyst.plans._
@@ -550,7 +551,7 @@ object UnsupportedOperationChecker extends Logging {
   }
 
   private def throwError(msg: String)(implicit operator: LogicalPlan): Nothing = {
-    throw new AnalysisException(
+    throw new ExtendedAnalysisException(
       msg, operator.origin.line, operator.origin.startPosition, Some(operator))
   }
 

sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala

@@ -23,7 +23,7 @@ import org.apache.hadoop.fs.Path
 
 import org.apache.spark.{SPARK_DOC_ROOT, SparkException, SparkThrowable, SparkThrowableHelper, SparkUnsupportedOperationException}
 import org.apache.spark.sql.AnalysisException
-import org.apache.spark.sql.catalyst.{FunctionIdentifier, QualifiedTableName, TableIdentifier}
+import org.apache.spark.sql.catalyst.{ExtendedAnalysisException, FunctionIdentifier, QualifiedTableName, TableIdentifier}
 import org.apache.spark.sql.catalyst.analysis.{CannotReplaceMissingTableException, FunctionAlreadyExistsException, NamespaceAlreadyExistsException, NoSuchFunctionException, NoSuchNamespaceException, NoSuchPartitionException, NoSuchTableException, ResolvedTable, Star, TableAlreadyExistsException, UnresolvedRegex}
 import org.apache.spark.sql.catalyst.catalog.{CatalogTable, InvalidUDFClassException}
 import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec
@@ -1895,7 +1895,7 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase {
 
   def streamJoinStreamWithoutEqualityPredicateUnsupportedError(plan: LogicalPlan): Throwable = {
     val errorClass = "_LEGACY_ERROR_TEMP_1181"
-    new AnalysisException(
+    new ExtendedAnalysisException(
       SparkThrowableHelper.getMessage(errorClass, Map.empty[String, String]),
       errorClass = Some(errorClass),
       messageParameters = Map.empty,

sql/core/src/test/resources/sql-tests/analyzer-results/ansi/array.sql.out

@@ -170,7 +170,7 @@ Project [sort_array(boolean_array#x, true) AS sort_array(boolean_array, true)#x,
 -- !query
 select sort_array(array('b', 'd'), '1')
 -- !query analysis
-org.apache.spark.sql.AnalysisException
+org.apache.spark.sql.catalyst.ExtendedAnalysisException
 {
   "errorClass" : "DATATYPE_MISMATCH.UNEXPECTED_INPUT_TYPE",
   "sqlState" : "42K09",
@@ -194,7 +194,7 @@ org.apache.spark.sql.AnalysisException
 -- !query
 select sort_array(array('b', 'd'), cast(NULL as boolean))
 -- !query analysis
-org.apache.spark.sql.AnalysisException
+org.apache.spark.sql.catalyst.ExtendedAnalysisException
 {
   "errorClass" : "DATATYPE_MISMATCH.UNEXPECTED_INPUT_TYPE",
   "sqlState" : "42K09",
@@ -353,7 +353,7 @@ Project [array_size(cast(null as array<void>)) AS array_size(NULL)#x]
 -- !query
 select array_size(map('a', 1, 'b', 2))
 -- !query analysis
-org.apache.spark.sql.AnalysisException
+org.apache.spark.sql.catalyst.ExtendedAnalysisException
 {
   "errorClass" : "DATATYPE_MISMATCH.UNEXPECTED_INPUT_TYPE",
   "sqlState" : "42K09",
@@ -475,7 +475,7 @@ Project [array_insert(array(1, 3, 4), -2, 2) AS array_insert(array(1, 3, 4), -2,
 -- !query
 select array_insert(array(1, 2, 3), 3, "4")
 -- !query analysis
-org.apache.spark.sql.AnalysisException
+org.apache.spark.sql.catalyst.ExtendedAnalysisException
 {
   "errorClass" : "DATATYPE_MISMATCH.ARRAY_FUNCTION_DIFF_TYPES",
   "sqlState" : "42K09",
@@ -534,7 +534,7 @@ Project [array_insert(array(2, 3, cast(null as int), 4), -5, 1) AS array_insert(
 -- !query
 select array_compact(id) from values (1) as t(id)
 -- !query analysis
-org.apache.spark.sql.AnalysisException
+org.apache.spark.sql.catalyst.ExtendedAnalysisException
 {
   "errorClass" : "DATATYPE_MISMATCH.UNEXPECTED_INPUT_TYPE",
   "sqlState" : "42K09",

sql/core/src/test/resources/sql-tests/analyzer-results/ansi/cast.sql.out

@@ -205,7 +205,7 @@ Project [hex(cast(abc as binary)) AS hex(CAST(abc AS BINARY))#x]
 -- !query
 SELECT HEX(CAST(CAST(123 AS byte) AS binary))
 -- !query analysis
-org.apache.spark.sql.AnalysisException
+org.apache.spark.sql.catalyst.ExtendedAnalysisException
 {
   "errorClass" : "DATATYPE_MISMATCH.CAST_WITH_CONF_SUGGESTION",
   "sqlState" : "42K09",
@@ -229,7 +229,7 @@ org.apache.spark.sql.AnalysisException
 -- !query
 SELECT HEX(CAST(CAST(-123 AS byte) AS binary))
 -- !query analysis
-org.apache.spark.sql.AnalysisException
+org.apache.spark.sql.catalyst.ExtendedAnalysisException
 {
   "errorClass" : "DATATYPE_MISMATCH.CAST_WITH_CONF_SUGGESTION",
   "sqlState" : "42K09",
@@ -253,7 +253,7 @@ org.apache.spark.sql.AnalysisException
 -- !query
 SELECT HEX(CAST(123S AS binary))
 -- !query analysis
-org.apache.spark.sql.AnalysisException
+org.apache.spark.sql.catalyst.ExtendedAnalysisException
 {
   "errorClass" : "DATATYPE_MISMATCH.CAST_WITH_CONF_SUGGESTION",
   "sqlState" : "42K09",
@@ -277,7 +277,7 @@ org.apache.spark.sql.AnalysisException
 -- !query
 SELECT HEX(CAST(-123S AS binary))
 -- !query analysis
-org.apache.spark.sql.AnalysisException
+org.apache.spark.sql.catalyst.ExtendedAnalysisException
 {
   "errorClass" : "DATATYPE_MISMATCH.CAST_WITH_CONF_SUGGESTION",
   "sqlState" : "42K09",
@@ -301,7 +301,7 @@ org.apache.spark.sql.AnalysisException
 -- !query
 SELECT HEX(CAST(123 AS binary))
 -- !query analysis
-org.apache.spark.sql.AnalysisException
+org.apache.spark.sql.catalyst.ExtendedAnalysisException
 {
   "errorClass" : "DATATYPE_MISMATCH.CAST_WITH_CONF_SUGGESTION",
   "sqlState" : "42K09",
@@ -325,7 +325,7 @@ org.apache.spark.sql.AnalysisException
 -- !query
 SELECT HEX(CAST(-123 AS binary))
 -- !query analysis
-org.apache.spark.sql.AnalysisException
+org.apache.spark.sql.catalyst.ExtendedAnalysisException
 {
   "errorClass" : "DATATYPE_MISMATCH.CAST_WITH_CONF_SUGGESTION",
   "sqlState" : "42K09",
@@ -349,7 +349,7 @@ org.apache.spark.sql.AnalysisException
 -- !query
 SELECT HEX(CAST(123L AS binary))
 -- !query analysis
-org.apache.spark.sql.AnalysisException
+org.apache.spark.sql.catalyst.ExtendedAnalysisException
 {
   "errorClass" : "DATATYPE_MISMATCH.CAST_WITH_CONF_SUGGESTION",
   "sqlState" : "42K09",
@@ -373,7 +373,7 @@ org.apache.spark.sql.AnalysisException
 -- !query
 SELECT HEX(CAST(-123L AS binary))
 -- !query analysis
-org.apache.spark.sql.AnalysisException
+org.apache.spark.sql.catalyst.ExtendedAnalysisException
 {
   "errorClass" : "DATATYPE_MISMATCH.CAST_WITH_CONF_SUGGESTION",
   "sqlState" : "42K09",