[SPARK-44531][CONNECT][SQL] Move encoder inference to sql/api

### What changes were proposed in this pull request?
This PR move encoder inference (ScalaReflection/RowEncoder/JavaTypeInference) into sql/api.

### Why are the changes needed?
We want to use encoder inference in the spark connect scala client. The client's dependency to catalyst is going away, so we need to move this.

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

### How was this patch tested?
Existing tests.

Closes #42134 from hvanhovell/SPARK-44531.

Authored-by: Herman van Hovell <[email protected]>
Signed-off-by: Herman van Hovell <[email protected]>

Author: hvanhovell

mllib/src/main/scala/org/apache/spark/ml/source/image/ImageFileFormat.scala

@@ -25,7 +25,7 @@ import org.apache.hadoop.mapreduce.Job
 import org.apache.spark.ml.image.ImageSchema
 import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.encoders.RowEncoder
+import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
 import org.apache.spark.sql.catalyst.expressions.UnsafeRow
 import org.apache.spark.sql.execution.datasources.{FileFormat, OutputWriterFactory, PartitionedFile}
 import org.apache.spark.sql.sources.{DataSourceRegister, Filter}
@@ -90,7 +90,7 @@ private[image] class ImageFileFormat extends FileFormat with DataSourceRegister
         if (requiredSchema.isEmpty) {
           filteredResult.map(_ => emptyUnsafeRow)
         } else {
-          val toRow = RowEncoder(requiredSchema).createSerializer()
+          val toRow = ExpressionEncoder(requiredSchema).createSerializer()
           filteredResult.map(row => toRow(row))
         }
       }

mllib/src/main/scala/org/apache/spark/ml/source/libsvm/LibSVMRelation.scala

@@ -31,7 +31,7 @@ import org.apache.spark.ml.linalg.{Vectors, VectorUDT}
 import org.apache.spark.mllib.util.MLUtils
 import org.apache.spark.sql.{Row, SparkSession}
 import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.encoders.RowEncoder
+import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
 import org.apache.spark.sql.catalyst.expressions.AttributeReference
 import org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection
 import org.apache.spark.sql.catalyst.types.DataTypeUtils
@@ -167,7 +167,7 @@ private[libsvm] class LibSVMFileFormat
             LabeledPoint(label, Vectors.sparse(numFeatures, indices, values))
           }
 
-      val toRow = RowEncoder(dataSchema).createSerializer()
+      val toRow = ExpressionEncoder(dataSchema).createSerializer()
       val fullOutput = dataSchema.map { f =>
         AttributeReference(f.name, f.dataType, f.nullable, f.metadata)()
       }

project/MimaExcludes.scala

@@ -40,6 +40,9 @@ object MimaExcludes {
 
   // Exclude rules for 3.5.x from 3.4.0
   lazy val v35excludes = defaultExcludes ++ Seq(
+    // [SPARK-44531][CONNECT][SQL] Move encoder inference to sql/api
+    ProblemFilters.exclude[MissingClassProblem]("org.apache.spark.sql.types.DataTypes"),
+    ProblemFilters.exclude[MissingClassProblem]("org.apache.spark.sql.types.SQLUserDefinedType"),
     // [SPARK-43165][SQL] Move canWrite to DataTypeUtils
     ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.sql.types.DataType.canWrite"),
     // [SPARK-43195][CORE] Remove unnecessary serializable wrapper in HadoopFSUtils

sql/api/pom.xml

@@ -35,6 +35,10 @@
     </properties>
 
     <dependencies>
+        <dependency>
+            <groupId>org.scala-lang</groupId>
+            <artifactId>scala-reflect</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.scala-lang.modules</groupId>
             <artifactId>scala-parser-combinators_${scala.binary.version}</artifactId>

sql/catalyst/src/main/java/org/apache/spark/sql/types/DataTypes.java renamed to sql/api/src/main/java/org/apache/spark/sql/types/DataTypes.java

@@ -41,6 +41,7 @@ private[sql] trait SqlApiConf {
   def timestampType: AtomicType
   def allowNegativeScaleOfDecimalEnabled: Boolean
   def charVarcharAsString: Boolean
+  def datetimeJava8ApiEnabled: Boolean
 }
 
 private[sql] object SqlApiConf {
@@ -76,4 +77,5 @@ private[sql] object DefaultSqlApiConf extends SqlApiConf {
   override def timestampType: AtomicType = TimestampType
   override def allowNegativeScaleOfDecimalEnabled: Boolean = false
   override def charVarcharAsString: Boolean = false
+  override def datetimeJava8ApiEnabled: Boolean = false
 }

sql/catalyst/src/main/java/org/apache/spark/sql/types/SQLUserDefinedType.java renamed to sql/api/src/main/java/org/apache/spark/sql/types/SQLUserDefinedType.java

@@ -24,10 +24,9 @@ import javax.annotation.Nonnull
 import scala.annotation.tailrec
 import scala.reflect.ClassTag
 
-import org.apache.spark.SPARK_DOC_ROOT
 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.QueryExecutionErrors
+import org.apache.spark.sql.errors.EncoderErrors
 import org.apache.spark.sql.types._
 
 /**
@@ -116,7 +115,7 @@ object JavaTypeInference {
 
     case c: Class[_] =>
       if (seenTypeSet.contains(c)) {
-        throw QueryExecutionErrors.cannotHaveCircularReferencesInBeanClassError(c)
+        throw EncoderErrors.cannotHaveCircularReferencesInBeanClassError(c)
       }
 
       // TODO: we should only collect properties that have getter and setter. However, some tests
@@ -139,7 +138,7 @@ object JavaTypeInference {
       JavaBeanEncoder(ClassTag(c), fields)
 
     case _ =>
-      throw QueryExecutionErrors.cannotFindEncoderForTypeError(t.toString, SPARK_DOC_ROOT)
+      throw EncoderErrors.cannotFindEncoderForTypeError(t.toString)
   }
 
   def getJavaBeanReadableProperties(beanClass: Class[_]): Array[PropertyDescriptor] = {
@@ -197,7 +196,7 @@ object JavaTypeInference {
             }
         }
       }
-      throw QueryExecutionErrors.unreachableError()
+      throw EncoderErrors.unreachableError()
     }
   }
 

sql/api/src/main/scala/org/apache/spark/sql/SqlApiConf.scala

@@ -26,12 +26,11 @@ import scala.util.{Failure, Success}
 
 import org.apache.commons.lang3.reflect.ConstructorUtils
 
-import org.apache.spark.SPARK_DOC_ROOT
 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.QueryExecutionErrors
+import org.apache.spark.sql.errors.EncoderErrors
 import org.apache.spark.sql.types._
 import org.apache.spark.unsafe.types.CalendarInterval
 
@@ -378,13 +377,13 @@ object ScalaReflection extends ScalaReflection {
 
       case t if definedByConstructorParams(t) =>
         if (seenTypeSet.contains(t)) {
-          throw QueryExecutionErrors.cannotHaveCircularReferencesInClassError(t.toString)
+          throw EncoderErrors.cannotHaveCircularReferencesInClassError(t.toString)
         }
         val params = getConstructorParameters(t).map {
           case (fieldName, fieldType) =>
             if (SourceVersion.isKeyword(fieldName) ||
               !SourceVersion.isIdentifier(encodeFieldNameToIdentifier(fieldName))) {
-              throw QueryExecutionErrors.cannotUseInvalidJavaIdentifierAsFieldNameError(
+              throw EncoderErrors.cannotUseInvalidJavaIdentifierAsFieldNameError(
                 fieldName,
                 path)
             }
@@ -397,7 +396,7 @@ object ScalaReflection extends ScalaReflection {
         }
         ProductEncoder(ClassTag(getClassFromType(t)), params)
       case _ =>
-        throw QueryExecutionErrors.cannotFindEncoderForTypeError(tpe.toString, SPARK_DOC_ROOT)
+        throw EncoderErrors.cannotFindEncoderForTypeError(tpe.toString)
     }
   }
 }
@@ -478,7 +477,7 @@ trait ScalaReflection extends Logging {
    */
   private def getCompanionConstructor(tpe: Type): Symbol = {
     def throwUnsupportedOperation = {
-      throw QueryExecutionErrors.cannotFindConstructorForTypeError(tpe.toString)
+      throw EncoderErrors.cannotFindConstructorForTypeError(tpe.toString)
     }
     tpe.typeSymbol.asClass.companion match {
       case NoSymbol => throwUnsupportedOperation
@@ -501,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 QueryExecutionErrors.primaryConstructorNotFoundError(tpe.getClass)
+        throw EncoderErrors.primaryConstructorNotFoundError(tpe.getClass)
       } else {
         primaryConstructorSymbol.get.asMethod.paramLists
       }

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

@@ -20,10 +20,9 @@ package org.apache.spark.sql.catalyst.encoders
 import scala.collection.mutable
 import scala.reflect.classTag
 
-import org.apache.spark.sql.Row
+import org.apache.spark.sql.{Row, SqlApiConf}
 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.QueryExecutionErrors
-import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.errors.EncoderErrors
 import org.apache.spark.sql.types._
 
 /**
@@ -59,14 +58,6 @@ import org.apache.spark.sql.types._
  * }}}
  */
 object RowEncoder {
-  def apply(schema: StructType, lenient: Boolean): ExpressionEncoder[Row] = {
-    ExpressionEncoder(encoderFor(schema, lenient))
-  }
-
-  def apply(schema: StructType): ExpressionEncoder[Row] = {
-    apply(schema, lenient = false)
-  }
-
   def encoderFor(schema: StructType): AgnosticEncoder[Row] = {
     encoderFor(schema, lenient = false)
   }
@@ -89,10 +80,10 @@ object RowEncoder {
     case dt: DecimalType => JavaDecimalEncoder(dt, lenientSerialization = true)
     case BinaryType => BinaryEncoder
     case StringType => StringEncoder
-    case TimestampType if SQLConf.get.datetimeJava8ApiEnabled => InstantEncoder(lenient)
+    case TimestampType if SqlApiConf.get.datetimeJava8ApiEnabled => InstantEncoder(lenient)
     case TimestampType => TimestampEncoder(lenient)
     case TimestampNTZType => LocalDateTimeEncoder
-    case DateType if SQLConf.get.datetimeJava8ApiEnabled => LocalDateEncoder(lenient)
+    case DateType if SqlApiConf.get.datetimeJava8ApiEnabled => LocalDateEncoder(lenient)
     case DateType => DateEncoder(lenient)
     case CalendarIntervalType => CalendarIntervalEncoder
     case _: DayTimeIntervalType => DayTimeIntervalEncoder
@@ -106,7 +97,7 @@ object RowEncoder {
         annotation.udt()
       } else {
         UDTRegistration.getUDTFor(udt.userClass.getName).getOrElse {
-          throw QueryExecutionErrors.userDefinedTypeNotAnnotatedAndRegisteredError(udt)
+          throw EncoderErrors.userDefinedTypeNotAnnotatedAndRegisteredError(udt)
         }
       }
       UDTEncoder(udt, udtClass.asInstanceOf[Class[_ <: UserDefinedType[_]]])

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

@@ -18,7 +18,7 @@ package org.apache.spark.sql.errors
 
 import java.util.Locale
 
-import org.apache.spark.QueryContext
+import org.apache.spark.{QueryContext, SparkRuntimeException}
 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}
@@ -73,4 +73,10 @@ 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/catalyst/src/main/scala/org/apache/spark/sql/catalyst/WalkedTypePath.scala renamed to sql/api/src/main/scala/org/apache/spark/sql/catalyst/WalkedTypePath.scala

@@ -0,0 +1,74 @@
+/*
+ * 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.errors
+
+import org.apache.spark.{SparkBuildInfo, SparkException, SparkRuntimeException, SparkUnsupportedOperationException}
+import org.apache.spark.sql.catalyst.WalkedTypePath
+import org.apache.spark.sql.types.UserDefinedType
+
+object EncoderErrors extends DataTypeErrorsBase {
+  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()))
+  }
+}

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

@@ -35,11 +35,6 @@
   </properties>
 
   <dependencies>
-    <dependency>
-      <groupId>org.scala-lang</groupId>
-      <artifactId>scala-reflect</artifactId>
-    </dependency>
-
     <dependency>
       <groupId>org.apache.spark</groupId>
       <artifactId>spark-core_${scala.binary.version}</artifactId>

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

@@ -20,7 +20,7 @@ package org.apache.spark.sql.catalyst.encoders
 import scala.reflect.ClassTag
 import scala.reflect.runtime.universe.TypeTag
 
-import org.apache.spark.sql.Encoder
+import org.apache.spark.sql.{Encoder, Row}
 import org.apache.spark.sql.catalyst.{DeserializerBuildHelper, InternalRow, JavaTypeInference, ScalaReflection, SerializerBuildHelper}
 import org.apache.spark.sql.catalyst.analysis.{Analyzer, GetColumnByOrdinal, SimpleAnalyzer, UnresolvedAttribute, UnresolvedExtractValue}
 import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder.{Deserializer, Serializer}
@@ -58,6 +58,12 @@ object ExpressionEncoder {
       enc.clsTag)
   }
 
+  def apply(schema: StructType): ExpressionEncoder[Row] = apply(schema, lenient = false)
+
+  def apply(schema: StructType, lenient: Boolean): ExpressionEncoder[Row] = {
+    apply(RowEncoder.encoderFor(schema, lenient))
+  }
+
   // TODO: improve error message for java bean encoder.
   def javaBean[T](beanClass: Class[T]): ExpressionEncoder[T] = {
      apply(JavaTypeInference.encoderFor(beanClass))

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

@@ -149,8 +149,8 @@ object MapPartitionsInR {
         broadcastVars,
         encoder.schema,
         schema,
-        CatalystSerde.generateObjAttr(RowEncoder(schema)),
-        deserialized))(RowEncoder(schema))
+        CatalystSerde.generateObjAttr(ExpressionEncoder(schema)),
+        deserialized))(ExpressionEncoder(schema))
     }
   }
 }
@@ -606,8 +606,8 @@ object FlatMapGroupsInR {
         UnresolvedDeserializer(valueDeserializer, dataAttributes),
         groupingAttributes,
         dataAttributes,
-        CatalystSerde.generateObjAttr(RowEncoder(schema)),
-        child))(RowEncoder(schema))
+        CatalystSerde.generateObjAttr(ExpressionEncoder(schema)),
+        child))(ExpressionEncoder(schema))
     }
   }
 }