init

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

@@ -24,6 +24,7 @@ import org.apache.avro.generic.GenericDatumReader
 import org.apache.avro.io.{BinaryDecoder, DecoderFactory}
 
 import org.apache.spark.SparkException
+import org.apache.spark.sql.avro.SchemaConverters
 import org.apache.spark.sql.catalyst.expressions.{ExpectsInputTypes, Expression, SpecificInternalRow, UnaryExpression}
 import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext, CodeGenerator, ExprCode}
 import org.apache.spark.sql.catalyst.util.{FailFastMode, ParseMode, PermissiveMode}

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

@@ -36,7 +36,8 @@ import org.apache.spark.util.ArrayImplicits._
  *                      the fields of the provided schema.
  * @param schema The required schema of records from datasource files.
  */
-abstract class StructFilters(pushedFilters: Seq[sources.Filter], schema: StructType) {
+abstract class StructFilters(
+    pushedFilters: Seq[sources.Filter], schema: StructType) extends Serializable {
 
   protected val filters = StructFilters.pushedFilters(pushedFilters.toArray, schema)
 

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -2204,6 +2204,16 @@ object SQLConf {
       .intConf
       .createWithDefault(3)
 
+  val STREAMING_STATE_STORE_ENCODING_FORMAT =
+    buildConf("spark.sql.streaming.stateStore.encodingFormat")
+      .doc("The encoding format used for stateful operators to store information" +
+        "in the state store")
+      .version("4.0.0")
+      .stringConf
+      .checkValue(v => Set("UnsafeRow", "Avro").contains(v),
+        "Valid versions are 'UnsafeRow' and 'Avro'")
+      .createWithDefault("UnsafeRow")
+
   // The feature is still in development, so it is still internal.
   val STATE_STORE_CHECKPOINT_FORMAT_VERSION =
     buildConf("spark.sql.streaming.stateStore.checkpointFormatVersion")

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

@@ -52,7 +52,7 @@ private[sql] class AvroDeserializer(
     filters: StructFilters,
     useStableIdForUnionType: Boolean,
     stableIdPrefixForUnionType: String,
-    recursiveFieldMaxDepth: Int) {
+    recursiveFieldMaxDepth: Int) extends Serializable {
 
   def this(
       rootAvroType: Schema,
@@ -463,7 +463,7 @@ private[sql] class AvroDeserializer(
    * A base interface for updating values inside catalyst data structure like `InternalRow` and
    * `ArrayData`.
    */
-  sealed trait CatalystDataUpdater {
+  sealed trait CatalystDataUpdater extends Serializable {
     def set(ordinal: Int, value: Any): Unit
 
     def setNullAt(ordinal: Int): Unit = set(ordinal, null)

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

@@ -47,7 +47,7 @@ private[sql] class AvroSerializer(
     rootAvroType: Schema,
     nullable: Boolean,
     positionalFieldMatch: Boolean,
-    datetimeRebaseMode: LegacyBehaviorPolicy.Value) extends Logging {
+    datetimeRebaseMode: LegacyBehaviorPolicy.Value) extends Logging with Serializable {
 
   def this(rootCatalystType: DataType, rootAvroType: Schema, nullable: Boolean) = {
     this(rootCatalystType, rootAvroType, nullable, positionalFieldMatch = false,