Simplification & Fixes

connector/avro/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala

@@ -37,8 +37,7 @@ import org.apache.spark.sql.catalyst.util.DateTimeConstants.MILLIS_PER_DAY
 import org.apache.spark.sql.catalyst.util.RebaseDateTime.RebaseSpec
 import org.apache.spark.sql.errors.QueryCompilationErrors
 import org.apache.spark.sql.execution.datasources.DataSourceUtils
-import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.internal.LegacyBehaviorPolicy
+import org.apache.spark.sql.internal.{LegacyBehaviorPolicy, SQLConf}
 import org.apache.spark.sql.types._
 import org.apache.spark.unsafe.types.UTF8String
 

connector/avro/src/main/scala/org/apache/spark/sql/avro/AvroOutputWriter.scala

@@ -32,8 +32,7 @@ import org.apache.spark.SPARK_VERSION_SHORT
 import org.apache.spark.sql.{SPARK_LEGACY_DATETIME_METADATA_KEY, SPARK_TIMEZONE_METADATA_KEY, SPARK_VERSION_METADATA_KEY}
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.execution.datasources.OutputWriter
-import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy
+import org.apache.spark.sql.internal.{LegacyBehaviorPolicy, SQLConf}
 import org.apache.spark.sql.types._
 
 // NOTE: This class is instantiated and used on executor side only, no need to be serializable.

connector/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala

@@ -37,8 +37,7 @@ import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.expressions.{SpecializedGetters, SpecificInternalRow}
 import org.apache.spark.sql.catalyst.util.DateTimeUtils
 import org.apache.spark.sql.execution.datasources.DataSourceUtils
-import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy
+import org.apache.spark.sql.internal.{LegacyBehaviorPolicy, SQLConf}
 import org.apache.spark.sql.types._
 
 /**

sql/api/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala

@@ -26,7 +26,7 @@ import scala.reflect.ClassTag
 
 import org.apache.spark.sql.catalyst.encoders.AgnosticEncoder
 import org.apache.spark.sql.catalyst.encoders.AgnosticEncoders.{ArrayEncoder, BinaryEncoder, BoxedBooleanEncoder, BoxedByteEncoder, BoxedDoubleEncoder, BoxedFloatEncoder, BoxedIntEncoder, BoxedLongEncoder, BoxedShortEncoder, DayTimeIntervalEncoder, DEFAULT_JAVA_DECIMAL_ENCODER, EncoderField, IterableEncoder, JavaBeanEncoder, JavaBigIntEncoder, JavaEnumEncoder, LocalDateTimeEncoder, MapEncoder, PrimitiveBooleanEncoder, PrimitiveByteEncoder, PrimitiveDoubleEncoder, PrimitiveFloatEncoder, PrimitiveIntEncoder, PrimitiveLongEncoder, PrimitiveShortEncoder, STRICT_DATE_ENCODER, STRICT_INSTANT_ENCODER, STRICT_LOCAL_DATE_ENCODER, STRICT_TIMESTAMP_ENCODER, StringEncoder, UDTEncoder, YearMonthIntervalEncoder}
-import org.apache.spark.sql.errors.EncoderErrors
+import org.apache.spark.sql.errors.ExecutionErrors
 import org.apache.spark.sql.types._
 
 /**
@@ -115,7 +115,7 @@ object JavaTypeInference {
 
     case c: Class[_] =>
       if (seenTypeSet.contains(c)) {
-        throw EncoderErrors.cannotHaveCircularReferencesInBeanClassError(c)
+        throw ExecutionErrors.cannotHaveCircularReferencesInBeanClassError(c)
       }
 
       // TODO: we should only collect properties that have getter and setter. However, some tests
@@ -138,7 +138,7 @@ object JavaTypeInference {
       JavaBeanEncoder(ClassTag(c), fields)
 
     case _ =>
-      throw EncoderErrors.cannotFindEncoderForTypeError(t.toString)
+      throw ExecutionErrors.cannotFindEncoderForTypeError(t.toString)
   }
 
   def getJavaBeanReadableProperties(beanClass: Class[_]): Array[PropertyDescriptor] = {
@@ -196,7 +196,7 @@ object JavaTypeInference {
             }
         }
       }
-      throw EncoderErrors.unreachableError()
+      throw ExecutionErrors.unreachableError()
     }
   }
 

sql/api/src/main/scala/org/apache/spark/sql/catalyst/ScalaReflection.scala

@@ -30,7 +30,7 @@ import org.apache.spark.internal.Logging
 import org.apache.spark.sql.Row
 import org.apache.spark.sql.catalyst.encoders.AgnosticEncoder
 import org.apache.spark.sql.catalyst.encoders.AgnosticEncoders._
-import org.apache.spark.sql.errors.EncoderErrors
+import org.apache.spark.sql.errors.ExecutionErrors
 import org.apache.spark.sql.types._
 import org.apache.spark.unsafe.types.CalendarInterval
 
@@ -377,13 +377,13 @@ object ScalaReflection extends ScalaReflection {
 
       case t if definedByConstructorParams(t) =>
         if (seenTypeSet.contains(t)) {
-          throw EncoderErrors.cannotHaveCircularReferencesInClassError(t.toString)
+          throw ExecutionErrors.cannotHaveCircularReferencesInClassError(t.toString)
         }
         val params = getConstructorParameters(t).map {
           case (fieldName, fieldType) =>
             if (SourceVersion.isKeyword(fieldName) ||
               !SourceVersion.isIdentifier(encodeFieldNameToIdentifier(fieldName))) {
-              throw EncoderErrors.cannotUseInvalidJavaIdentifierAsFieldNameError(
+              throw ExecutionErrors.cannotUseInvalidJavaIdentifierAsFieldNameError(
                 fieldName,
                 path)
             }
@@ -396,7 +396,7 @@ object ScalaReflection extends ScalaReflection {
         }
         ProductEncoder(ClassTag(getClassFromType(t)), params)
       case _ =>
-        throw EncoderErrors.cannotFindEncoderForTypeError(tpe.toString)
+        throw ExecutionErrors.cannotFindEncoderForTypeError(tpe.toString)
     }
   }
 }
@@ -477,7 +477,7 @@ trait ScalaReflection extends Logging {
    */
   private def getCompanionConstructor(tpe: Type): Symbol = {
     def throwUnsupportedOperation = {
-      throw EncoderErrors.cannotFindConstructorForTypeError(tpe.toString)
+      throw ExecutionErrors.cannotFindConstructorForTypeError(tpe.toString)
     }
     tpe.typeSymbol.asClass.companion match {
       case NoSymbol => throwUnsupportedOperation
@@ -500,7 +500,7 @@ trait ScalaReflection extends Logging {
       val primaryConstructorSymbol: Option[Symbol] = constructorSymbol.asTerm.alternatives.find(
         s => s.isMethod && s.asMethod.isPrimaryConstructor)
       if (primaryConstructorSymbol.isEmpty) {
-        throw EncoderErrors.primaryConstructorNotFoundError(tpe.getClass)
+        throw ExecutionErrors.primaryConstructorNotFoundError(tpe.getClass)
       } else {
         primaryConstructorSymbol.get.asMethod.paramLists
       }

sql/api/src/main/scala/org/apache/spark/sql/catalyst/encoders/RowEncoder.scala

@@ -20,9 +20,10 @@ package org.apache.spark.sql.catalyst.encoders
 import scala.collection.mutable
 import scala.reflect.classTag
 
-import org.apache.spark.sql.{Row, SqlApiConf}
+import org.apache.spark.sql.Row
 import org.apache.spark.sql.catalyst.encoders.AgnosticEncoders.{BinaryEncoder, BoxedBooleanEncoder, BoxedByteEncoder, BoxedDoubleEncoder, BoxedFloatEncoder, BoxedIntEncoder, BoxedLongEncoder, BoxedShortEncoder, CalendarIntervalEncoder, DateEncoder, DayTimeIntervalEncoder, EncoderField, InstantEncoder, IterableEncoder, JavaDecimalEncoder, LocalDateEncoder, LocalDateTimeEncoder, MapEncoder, NullEncoder, RowEncoder => AgnosticRowEncoder, StringEncoder, TimestampEncoder, UDTEncoder, YearMonthIntervalEncoder}
-import org.apache.spark.sql.errors.EncoderErrors
+import org.apache.spark.sql.errors.ExecutionErrors
+import org.apache.spark.sql.internal.SqlApiConf
 import org.apache.spark.sql.types._
 
 /**
@@ -97,7 +98,7 @@ object RowEncoder {
         annotation.udt()
       } else {
         UDTRegistration.getUDTFor(udt.userClass.getName).getOrElse {
-          throw EncoderErrors.userDefinedTypeNotAnnotatedAndRegisteredError(udt)
+          throw ExecutionErrors.userDefinedTypeNotAnnotatedAndRegisteredError(udt)
         }
       }
       UDTEncoder(udt, udtClass.asInstanceOf[Class[_ <: UserDefinedType[_]]])

sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/DateTimeFormatterHelper.scala

@@ -24,7 +24,6 @@ import java.time.temporal.{ChronoField, TemporalAccessor, TemporalQueries}
 import java.util
 import java.util.{Collections, Date, Locale}
 
-import org.apache.spark.SparkBuildInfo.{spark_doc_root => SPARK_DOC_ROOT}
 import org.apache.spark.sql.catalyst.util.DateTimeFormatterHelper._
 import org.apache.spark.sql.errors.ExecutionErrors
 import org.apache.spark.sql.internal.LegacyBehaviorPolicy._
@@ -179,13 +178,12 @@ trait DateTimeFormatterHelper {
       } catch {
         case _: Throwable => throw e
       }
-      throw ExecutionErrors.failToRecognizePatternAfterUpgradeError(
-        pattern, e, SPARK_DOC_ROOT)
+      throw ExecutionErrors.failToRecognizePatternAfterUpgradeError(pattern, e)
   }
 
   protected def checkInvalidPattern(pattern: String): PartialFunction[Throwable, Nothing] = {
     case e: IllegalArgumentException =>
-      throw ExecutionErrors.failToRecognizePatternError(pattern, e, SPARK_DOC_ROOT)
+      throw ExecutionErrors.failToRecognizePatternError(pattern, e)
   }
 }
 

sql/api/src/main/scala/org/apache/spark/sql/errors/DataTypeErrorsBase.scala

@@ -18,7 +18,7 @@ package org.apache.spark.sql.errors
 
 import java.util.Locale
 
-import org.apache.spark.{QueryContext, SparkRuntimeException}
+import org.apache.spark.QueryContext
 import org.apache.spark.sql.catalyst.trees.SQLQueryContext
 import org.apache.spark.sql.catalyst.util.{AttributeNameParser, QuotingUtils, SparkStringUtils}
 import org.apache.spark.sql.types.{AbstractDataType, DataType, TypeCollection}
@@ -96,10 +96,4 @@ private[sql] trait DataTypeErrorsBase {
   def getQueryContext(sqlContext: SQLQueryContext): Array[QueryContext] = {
     if (sqlContext == null) Array.empty else Array(sqlContext.asInstanceOf[QueryContext])
   }
-
-  def unreachableError(err: String = ""): SparkRuntimeException = {
-    new SparkRuntimeException(
-      errorClass = "_LEGACY_ERROR_TEMP_2028",
-      messageParameters = Map("err" -> err))
-  }
 }

sql/api/src/main/scala/org/apache/spark/sql/errors/ExecutionErrors.scala

@@ -21,11 +21,12 @@ import java.time.temporal.ChronoField
 
 import org.apache.arrow.vector.types.pojo.ArrowType
 
-import org.apache.spark.{SparkArithmeticException, SparkDateTimeException, SparkIllegalArgumentException, SparkRuntimeException, SparkUnsupportedOperationException, SparkUpgradeException}
+import org.apache.spark.{SparkArithmeticException, SparkBuildInfo, SparkDateTimeException, SparkException, SparkIllegalArgumentException, SparkRuntimeException, SparkUnsupportedOperationException, SparkUpgradeException}
+import org.apache.spark.sql.catalyst.WalkedTypePath
 import org.apache.spark.sql.catalyst.trees.SQLQueryContext
 import org.apache.spark.sql.catalyst.util.DateTimeUtils
 import org.apache.spark.sql.internal.SqlApiConf
-import org.apache.spark.sql.types.{DataType, DoubleType, FloatType, LongType, StringType}
+import org.apache.spark.sql.types.{DataType, DoubleType, FloatType, LongType, StringType, UserDefinedType}
 import org.apache.spark.unsafe.types.UTF8String
 
 private[sql] trait ExecutionErrors extends DataTypeErrorsBase {
@@ -45,7 +46,7 @@ private[sql] trait ExecutionErrors extends DataTypeErrorsBase {
       summary = "")
   }
 
-  def failToParseDateTimeInNewParserError(s: String, e: Throwable): Throwable = {
+  def failToParseDateTimeInNewParserError(s: String, e: Throwable): SparkUpgradeException = {
     new SparkUpgradeException(
       errorClass = "INCONSISTENT_BEHAVIOR_CROSS_VERSION.PARSE_DATETIME_BY_NEW_PARSER",
       messageParameters = Map(
@@ -55,23 +56,22 @@ private[sql] trait ExecutionErrors extends DataTypeErrorsBase {
   }
 
   def failToRecognizePatternAfterUpgradeError(
-      pattern: String, e: Throwable, docroot: String): Throwable = {
+      pattern: String, e: Throwable): SparkUpgradeException = {
     new SparkUpgradeException(
       errorClass = "INCONSISTENT_BEHAVIOR_CROSS_VERSION.DATETIME_PATTERN_RECOGNITION",
       messageParameters = Map(
         "pattern" -> toSQLValue(pattern),
         "config" -> toSQLConf(SqlApiConf.LEGACY_TIME_PARSER_POLICY_KEY),
-        "docroot" -> docroot),
+        "docroot" -> SparkBuildInfo.spark_doc_root),
       e)
   }
 
-  def failToRecognizePatternError(
-      pattern: String, e: Throwable, docroot: String): SparkRuntimeException = {
+  def failToRecognizePatternError(pattern: String, e: Throwable): SparkRuntimeException = {
     new SparkRuntimeException(
       errorClass = "_LEGACY_ERROR_TEMP_2130",
       messageParameters = Map(
         "pattern" -> toSQLValue(pattern),
-        "docroot" -> docroot),
+        "docroot" -> SparkBuildInfo.spark_doc_root),
       cause = e)
   }
 
@@ -219,6 +219,57 @@ private[sql] trait ExecutionErrors extends DataTypeErrorsBase {
       errorClass = "UNSUPPORTED_DATATYPE",
       messageParameters = Map("typeName" -> toSQLType(typeName)))
   }
+
+  def userDefinedTypeNotAnnotatedAndRegisteredError(udt: UserDefinedType[_]): Throwable = {
+    new SparkException(
+      errorClass = "_LEGACY_ERROR_TEMP_2155",
+      messageParameters = Map(
+        "userClass" -> udt.userClass.getName),
+      cause = null)
+  }
+
+  def cannotFindEncoderForTypeError(typeName: String): SparkUnsupportedOperationException = {
+    new SparkUnsupportedOperationException(
+      errorClass = "ENCODER_NOT_FOUND",
+      messageParameters = Map(
+        "typeName" -> typeName,
+        "docroot" -> SparkBuildInfo.spark_doc_root))
+  }
+
+  def cannotHaveCircularReferencesInBeanClassError(
+      clazz: Class[_]): SparkUnsupportedOperationException = {
+    new SparkUnsupportedOperationException(
+      errorClass = "_LEGACY_ERROR_TEMP_2138",
+      messageParameters = Map("clazz" -> clazz.toString))
+  }
+
+  def cannotFindConstructorForTypeError(tpe: String): SparkUnsupportedOperationException = {
+    new SparkUnsupportedOperationException(
+      errorClass = "_LEGACY_ERROR_TEMP_2144",
+      messageParameters = Map(
+        "tpe" -> tpe))
+  }
+
+  def cannotHaveCircularReferencesInClassError(t: String): SparkUnsupportedOperationException = {
+    new SparkUnsupportedOperationException(
+      errorClass = "_LEGACY_ERROR_TEMP_2139",
+      messageParameters = Map("t" -> t))
+  }
+
+  def cannotUseInvalidJavaIdentifierAsFieldNameError(
+      fieldName: String, walkedTypePath: WalkedTypePath): SparkUnsupportedOperationException = {
+    new SparkUnsupportedOperationException(
+      errorClass = "_LEGACY_ERROR_TEMP_2140",
+      messageParameters = Map(
+        "fieldName" -> fieldName,
+        "walkedTypePath" -> walkedTypePath.toString))
+  }
+
+  def primaryConstructorNotFoundError(cls: Class[_]): SparkRuntimeException = {
+    new SparkRuntimeException(
+      errorClass = "_LEGACY_ERROR_TEMP_2021",
+      messageParameters = Map("cls" -> cls.toString))
+  }
 }
 
 private[sql] object ExecutionErrors extends ExecutionErrors