[SPARK-26836][SQL] Supporting Avro schema evolution for partitioned Hive tables with "avro.schema.literal"

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

@@ -905,6 +905,18 @@ object SQLConf {
     .checkValues(HiveCaseSensitiveInferenceMode.values.map(_.toString))
     .createWithDefault(HiveCaseSensitiveInferenceMode.NEVER_INFER.toString)
 
+  val HIVE_AVRO_SCHEMA_EVOLUTION_ENABLED =
+    buildConf("spark.sql.hive.avroSchemaEvolution.enabled")
+      .internal()
+      .doc("When true, enable Avro schema evolution support for partitioned Hive table by " +
+        "giving priority to Avro related table properties defined at table level over the " +
+        "the ones given at partition level. Thus when an evolved schema is set for the table " +
+        "this schema will be used for reading the partition data and not the original Avro " +
+        "schema which was used for writing the partition.")
+      .version("3.2.0")
+      .booleanConf
+      .createWithDefault(false)
+
   val HIVE_TABLE_PROPERTY_LENGTH_THRESHOLD =
     buildConf("spark.sql.hive.tablePropertyLengthThreshold")
       .internal()
@@ -3307,6 +3319,8 @@ class SQLConf extends Serializable with Logging {
   def caseSensitiveInferenceMode: HiveCaseSensitiveInferenceMode.Value =
     HiveCaseSensitiveInferenceMode.withName(getConf(HIVE_CASE_SENSITIVE_INFERENCE))
 
+  def avroSchemaEvolutionEnabled: Boolean = getConf(HIVE_AVRO_SCHEMA_EVOLUTION_ENABLED)
+
   def gatherFastStats: Boolean = getConf(GATHER_FASTSTAT)
 
   def optimizerMetadataOnly: Boolean = getConf(OPTIMIZER_METADATA_ONLY)

sql/hive/src/main/scala/org/apache/spark/sql/hive/TableReader.scala

@@ -28,6 +28,7 @@ import org.apache.hadoop.hive.ql.exec.Utilities
 import org.apache.hadoop.hive.ql.metadata.{Partition => HivePartition, Table => HiveTable}
 import org.apache.hadoop.hive.ql.plan.TableDesc
 import org.apache.hadoop.hive.serde2.Deserializer
+import org.apache.hadoop.hive.serde2.avro.AvroSerdeUtils.AvroTableProperties
 import org.apache.hadoop.hive.serde2.objectinspector.{ObjectInspectorConverters, StructObjectInspector}
 import org.apache.hadoop.hive.serde2.objectinspector.primitive._
 import org.apache.hadoop.io.Writable
@@ -239,7 +240,8 @@ class HadoopTableReader(
       fillPartitionKeys(partValues, mutableRow)
 
       val tableProperties = tableDesc.getProperties
-
+      val avroTablePropertyKeys = HadoopTableReader.avroTableProperties
+      val avroSchemaEvolutionEnabled = conf.avroSchemaEvolutionEnabled
       // Create local references so that the outer object isn't serialized.
       val localTableDesc = tableDesc
       createHadoopRDD(localTableDesc, inputPathStr).mapPartitions { iter =>
@@ -248,11 +250,17 @@ class HadoopTableReader(
         // SPARK-13709: For SerDes like AvroSerDe, some essential information (e.g. Avro schema
         // information) may be defined in table properties. Here we should merge table properties
         // and partition properties before initializing the deserializer. Note that partition
-        // properties take a higher priority here. For example, a partition may have a different
-        // SerDe as the one defined in table properties.
+        // properties take a higher priority here except for the Avro table properties when
+        // "spark.sql.hive.avroSchemaEvolution.enabled" is set because in that case the properties
+        // given at table level will be used (for details please check SPARK-26836).
+        // For example, a partition may have a different SerDe as the one defined in table
+        // properties.
         val props = new Properties(tableProperties)
-        partProps.asScala.foreach {
-          case (key, value) => props.setProperty(key, value)
+        partProps.asScala.foreach { case (key, value) =>
+          if (!avroSchemaEvolutionEnabled ||
+            !avroTablePropertyKeys.contains(key) || !tableProperties.containsKey(key)) {
+            props.setProperty(key, value)
+          }
         }
         DeserializerLock.synchronized {
           deserializer.initialize(hconf, props)
@@ -388,6 +396,9 @@ private[hive] object HiveTableUtil {
 private[hive] object DeserializerLock
 
 private[hive] object HadoopTableReader extends HiveInspectors with Logging {
+
+  val avroTableProperties = AvroTableProperties.values().map(_.getPropName()).toSet
+
   /**
    * Curried. After given an argument for 'path', the resulting JobConf => Unit closure is used to
    * instantiate a HadoopRDD.

sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala

@@ -1883,6 +1883,60 @@ class HiveDDLSuite
     }
   }
 
+  test("SPARK-26836: support Avro schema evolution") {
+    withTable("t") {
+      val originalSchema =
+        """
+          |{
+          |  "namespace": "test",
+          |  "name": "some_schema",
+          |  "type": "record",
+          |  "fields": [
+          |    {
+          |      "name": "col2",
+          |      "type": "string"
+          |    }
+          |  ]
+          |}
+        """.stripMargin
+      sql(
+        s"""
+           |CREATE TABLE t PARTITIONED BY (ds string)
+           |ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.avro.AvroSerDe'
+           |WITH SERDEPROPERTIES ('avro.schema.literal'='$originalSchema')
+           |STORED AS
+           |INPUTFORMAT 'org.apache.hadoop.hive.ql.io.avro.AvroContainerInputFormat'
+           |OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.avro.AvroContainerOutputFormat'
+        """.stripMargin)
+      sql(s"INSERT INTO t partition (ds='1981-01-07') VALUES ('col2_value')")
+      val evolvedSchema =
+        """
+          |{
+          |  "namespace": "test",
+          |  "name": "some_schema",
+          |  "type": "record",
+          |  "fields": [
+          |    {
+          |      "name": "col1",
+          |      "type": "string",
+          |      "default": "col1_default"
+          |    },
+          |    {
+          |      "name": "col2",
+          |      "type": "string"
+          |    }
+          |  ]
+          |}
+        """.stripMargin
+      sql(s"""ALTER TABLE t SET SERDEPROPERTIES ('avro.schema.literal'='$evolvedSchema')""")
+      sql(s"INSERT INTO t partition (ds='1983-04-27') VALUES ('col1_value', 'col2_value')")
+      withSQLConf("spark.sql.hive.avroSchemaEvolution.enabled" -> "true") {
+        checkAnswer(spark.table("t"), Row("col1_default", "col2_value", "1981-01-07")
+          :: Row("col1_value", "col2_value", "1983-04-27") :: Nil)
+      }
+    }
+  }
+
   test("append data to hive serde table") {
     withTable("t", "t1") {
       Seq(1 -> "a").toDF("i", "j")