Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
attilapiros wants to merge 10 commits into from
Closed

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
a5f8d1a
Initial commit.
attilapiros Jan 11, 2021
9cba254
applying review comments
attilapiros Jan 12, 2021
080c29b
Merge branch 'master' into SPARK-26836
attilapiros Jan 12, 2021
232d8c0
simplify condition about avro properties
attilapiros Jan 15, 2021
009ecc0
remove spark.sql.hive.avroSchemaEvolution.enabled
attilapiros Jan 16, 2021
8dc2c08
fix indentation
attilapiros Jan 16, 2021
6892f31
remove AvroTableProperties.*
attilapiros Feb 4, 2021
28140f5
Narrowing down the scope to avro.schema.literal
attilapiros Feb 4, 2021
92abaa3
apply review comments
attilapiros Feb 4, 2021
20dc167
Fix indent + remove column test
attilapiros Feb 5, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 9 additions & 3 deletions sql/hive/src/main/scala/org/apache/spark/sql/hive/TableReader.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -248,10 +249,15 @@ 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
// to support schema evolution: 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 {
partProps.asScala.filterNot { case (k, _) =>
k == AvroTableProperties.SCHEMA_LITERAL.getPropName() && tableProperties.containsKey(k)
}.foreach {
case (key, value) => props.setProperty(key, value)
}
DeserializerLock.synchronized {
Expand Down
104 changes: 104 additions & 0 deletions sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1883,6 +1883,110 @@ class HiveDDLSuite
}
}

test("SPARK-26836: support Avro schema evolution (add column)") {
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("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("INSERT INTO t partition (ds='1983-04-27') VALUES ('col1_value', 'col2_value')")
checkAnswer(spark.table("t"), Row("col1_default", "col2_value", "1981-01-07")
:: Row("col1_value", "col2_value", "1983-04-27") :: Nil)
}
}

test("SPARK-26836: support Avro schema evolution (remove column)") {
withTable("t") {
val originalSchema =
"""
|{
| "namespace": "test",
| "name": "some_schema",
| "type": "record",
| "fields": [
| {
| "name": "col1",
| "type": "string",
| "default": "col1_default"
| },
| {
| "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("INSERT INTO t partition (ds='1983-04-27') VALUES ('col1_value', 'col2_value')")
val evolvedSchema =
"""
|{
| "namespace": "test",
| "name": "some_schema",
| "type": "record",
| "fields": [
| {
| "name": "col2",
| "type": "string"
| }
| ]
|}
""".stripMargin
sql(s"ALTER TABLE t SET SERDEPROPERTIES ('avro.schema.literal'='$evolvedSchema')")
sql("INSERT INTO t partition (ds='1981-01-07') VALUES ('col2_value')")
checkAnswer(spark.table("t"), Row("col2_value", "1981-01-07")
:: Row("col2_value", "1983-04-27") :: Nil)
}
}

test("append data to hive serde table") {
withTable("t", "t1") {
Seq(1 -> "a").toDF("i", "j")
Expand Down