You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by lindblombr <gi...@git.apache.org> on 2018/07/23 16:13:52 UTC
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL][WIP]: Built-in AVRO ...
GitHub user lindblombr opened a pull request:
https://github.com/apache/spark/pull/21847
[SPARK-24855][SQL][EXTERNAL][WIP]: Built-in AVRO support should support specified schema on write
## What changes were proposed in this pull request?
Allows `avroSchema` option to be specified on write, allowing a user to specify a schema in cases where this is required. A trivial use case is reading in an avro dataset, making some small adjustment to a column or columns and writing out using the same schema. Implicit schema creation from SQL Struct results in a schema that while for the most part, is functionally similar, is not necessarily compatible.
Allows `fixed` Field type to be utilized for records of specified `avroSchema`
## How was this patch tested?
Unit tests in AvroSuite are extended to write out included sample datasets using new `avroSchema` option and verify readability of records. Ideally, tests should be added for each `Schema.Type` to ensure records are written with full fidelity.
Please review http://spark.apache.org/contributing.html before opening a pull request.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/lindblombr/spark specify_schema_on_write
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/21847.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #21847
----
commit 30fc1ae98a2c29824b83e956c2cbe87e9bbb3749
Author: Brian Lindblom <bl...@...>
Date: 2018-07-21T20:43:26Z
SPARK-24855: Built-in AVRO support should support specified schema on write
----
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Merged build finished. Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by dbtsai <gi...@git.apache.org>.
Github user dbtsai commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r206359706
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -165,16 +182,118 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
result
}
- private def resolveNullableType(avroType: Schema, nullable: Boolean): Schema = {
- if (nullable) {
+ // Resolve an Avro union against a supplied DataType, i.e. a LongType compared against
+ // a ["null", "long"] should return a schema of type Schema.Type.LONG
+ // This function also handles resolving a DataType against unions of 2 or more types, i.e.
+ // an IntType resolves against a ["int", "long", "null"] will correctly return a schema of
+ // type Schema.Type.LONG
+ private def resolveUnionType(avroType: Schema, catalystType: DataType,
+ nullable: Boolean): Schema = {
--- End diff --
Two spaces after a new line in a fun def.
```scala
private def resolveUnionType(avroType: Schema, catalystType: DataType,
nullable: Boolean): Schema = {
```
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by dbtsai <gi...@git.apache.org>.
Github user dbtsai commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r205648911
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -87,17 +88,33 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
case d: DecimalType =>
(getter, ordinal) => getter.getDecimal(ordinal, d.precision, d.scale).toString
case StringType =>
- (getter, ordinal) => new Utf8(getter.getUTF8String(ordinal).getBytes)
+ (getter, ordinal) =>
+ if (avroType.getType == Type.ENUM) {
+ new GenericData.EnumSymbol(avroType, getter.getUTF8String(ordinal).toString)
+ } else {
+ new Utf8(getter.getUTF8String(ordinal).getBytes)
+ }
case BinaryType =>
- (getter, ordinal) => ByteBuffer.wrap(getter.getBinary(ordinal))
+ (getter, ordinal) =>
+ val data = getter.getBinary(ordinal)
+ if (avroType.getType == Type.FIXED) {
+ // Handles fixed-type fields in output schema. Test case is included in test.avro
+ // as it includes several fixed fields that would fail if we specify schema
+ // on-write without this condition
+ val fixed = new GenericData.Fixed(avroType)
+ fixed.bytes(data)
+ fixed
+ } else {
+ ByteBuffer.wrap(data)
+ }
--- End diff --
This might be slow. In the executors, when each row is going to be serialized, the whole `if-else` will be executed again and agin to get a specialized converter. We can consider to resolve the specialized types earlier in driver by
```scala
import org.apache.avro.generic.GenericData.{Fixed, EnumSymbol}
...
case StringType =>
if (avroType.getType == Type.ENUM) {
(getter, ordinal) => new EnumSymbol(avroType, getter.getUTF8String(ordinal).toString)
} else {
(getter, ordinal) => new Utf8(getter.getUTF8String(ordinal).getBytes)
}
case BinaryType =>
if (avroType.getType == Type.FIXED) {
(getter, ordinal) => new Fixed(avroType, getter.getBinary(ordinal))
} else {
(getter, ordinal) => ByteBuffer.wrap(getter.getBinary(ordinal))
}
```
so the returned lambda expression will not have any check on `FIXED` or `ENUM` types.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r205550412
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -165,16 +183,100 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
result
}
- private def resolveNullableType(avroType: Schema, nullable: Boolean): Schema = {
- if (nullable) {
+ private def resolveNullableType(avroType: Schema, catalystType: DataType,
--- End diff --
Not caused by this PR, but we better explain what this method does. Can you add a comment for it?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r209116303
--- Diff: external/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala ---
@@ -725,6 +744,205 @@ class AvroSuite extends QueryTest with SharedSQLContext with SQLTestUtils {
assert(result === Row("foo"))
}
+ test("support user provided avro schema for writing nullable enum type") {
+ withTempPath { tempDir =>
+ val avroSchema =
+ """
+ |{
+ | "type" : "record",
+ | "name" : "test_schema",
+ | "fields" : [{
+ | "name": "enum",
+ | "type": [{ "type": "enum",
+ | "name": "Suit",
+ | "symbols" : ["SPADES", "HEARTS", "DIAMONDS", "CLUBS"]
+ | }, "null"]
+ | }]
+ |}
+ """.stripMargin
+
+ val df = spark.createDataFrame(spark.sparkContext.parallelize(Seq(
+ Row("SPADES"), Row(null), Row("HEARTS"), Row("DIAMONDS"),
+ Row(null), Row("CLUBS"), Row("HEARTS"), Row("SPADES"))),
+ StructType(Seq(StructField("Suit", StringType, true))))
+
+ val tempSaveDir = s"$tempDir/save/"
+
+ df.write.format("avro").option("avroSchema", avroSchema).save(tempSaveDir)
+
+ checkAnswer(df, spark.read.format("avro").load(tempSaveDir))
+ checkAvroSchemaEquals(avroSchema, getAvroSchemaStringFromFiles(tempSaveDir))
+
+ // Writing df containing data not in the enum will throw an exception
+ val message = intercept[SparkException] {
+ spark.createDataFrame(spark.sparkContext.parallelize(Seq(
+ Row("SPADES"), Row("NOT-IN-ENUM"), Row("HEARTS"), Row("DIAMONDS"))),
+ StructType(Seq(StructField("Suit", StringType, true))))
+ .write.format("avro").option("avroSchema", avroSchema)
+ .save(s"$tempDir/${UUID.randomUUID()}")
+ }.getCause.getMessage
+ assert(message.contains("org.apache.spark.sql.avro.IncompatibleSchemaException: " +
+ "Cannot write \"NOT-IN-ENUM\" since it's not defined in enum"))
+ }
+ }
+
+ test("support user provided avro schema for writing non-nullable enum type") {
+ withTempPath { tempDir =>
+ val avroSchema =
+ """
+ |{
+ | "type" : "record",
+ | "name" : "test_schema",
+ | "fields" : [{
+ | "name": "enum",
+ | "type": { "type": "enum",
+ | "name": "Suit",
+ | "symbols" : ["SPADES", "HEARTS", "DIAMONDS", "CLUBS"]
+ | }
+ | }]
+ |}
+ """.stripMargin
+
+ val dfWithNull = spark.createDataFrame(spark.sparkContext.parallelize(Seq(
+ Row("SPADES"), Row(null), Row("HEARTS"), Row("DIAMONDS"),
+ Row(null), Row("CLUBS"), Row("HEARTS"), Row("SPADES"))),
+ StructType(Seq(StructField("Suit", StringType, true))))
+
+ val df = spark.createDataFrame(dfWithNull.na.drop().rdd,
+ StructType(Seq(StructField("Suit", StringType, false))))
+
+ val tempSaveDir = s"$tempDir/save1/"
--- End diff --
nit: seems can still use `save`?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL][WIP]: Built-in AVRO support...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/93565/
Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL][WIP]: Built-in AVRO support...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #93604 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93604/testReport)** for PR 21847 at commit [`6f686d6`](https://github.com/apache/spark/commit/6f686d6401fa991f6bcb0613bd9c4256cc7cffb9).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by lindblombr <gi...@git.apache.org>.
Github user lindblombr commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r205591619
--- Diff: external/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala ---
@@ -40,12 +41,40 @@ import org.apache.spark.sql.types._
class AvroSuite extends QueryTest with SharedSQLContext with SQLTestUtils {
val episodesAvro = testFile("episodes.avro")
val testAvro = testFile("test.avro")
+ val messyAvro = testFile("messy.avro")
+ val multiRecordTypeUnionAvro = testFile("multirecordtypeunion.avro")
+ val episodesSchemaFile = testFile("episodes.avsc")
+ val testSchemaFile = testFile("test.avsc")
+ val messySchemaFile = testFile("messy.avsc")
+ val multiRecordTypeUnionSchemaFile = testFile("multirecordtypeunion.avsc")
override protected def beforeAll(): Unit = {
super.beforeAll()
spark.conf.set("spark.sql.files.maxPartitionBytes", 1024)
}
+ def forceSchemaCheck(spark: SparkSession, inputPath: String, schemaFile: String): Unit = {
--- End diff --
Fixed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r206368071
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -165,16 +182,118 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
result
}
- private def resolveNullableType(avroType: Schema, nullable: Boolean): Schema = {
- if (nullable) {
+ // Resolve an Avro union against a supplied DataType, i.e. a LongType compared against
+ // a ["null", "long"] should return a schema of type Schema.Type.LONG
+ // This function also handles resolving a DataType against unions of 2 or more types, i.e.
+ // an IntType resolves against a ["int", "long", "null"] will correctly return a schema of
+ // type Schema.Type.LONG
+ private def resolveUnionType(avroType: Schema, catalystType: DataType,
+ nullable: Boolean): Schema = {
+ if (avroType.getType == Type.UNION) {
// avro uses union to represent nullable type.
- val fields = avroType.getTypes.asScala
- assert(fields.length == 2)
- val actualType = fields.filter(_.getType != NULL)
- assert(actualType.length == 1)
+ val fieldTypes = avroType.getTypes.asScala
+
+ // If we're nullable, we need to have at least two types. Cases with more than two types
+ // are captured in test("read read-write, read-write w/ schema, read") w/ test.avro input
+ if (nullable && fieldTypes.length < 2) {
+ throw new IncompatibleSchemaException(
+ s"Cannot resolve nullable ${catalystType} against union type ${avroType}")
+ }
+
+ val actualType = catalystType match {
+ case NullType => fieldTypes.filter(_.getType == Type.NULL)
+ case BooleanType => fieldTypes.filter(_.getType == Type.BOOLEAN)
+ case ByteType => fieldTypes.filter(_.getType == Type.INT)
+ case BinaryType =>
+ val at = fieldTypes.filter(x => x.getType == Type.BYTES || x.getType == Type.FIXED)
+ if (at.length > 1) {
+ throw new IncompatibleSchemaException(
+ s"Cannot resolve schema of ${catalystType} against union ${avroType.toString}")
+ } else {
+ at
+ }
+ case ShortType | IntegerType => fieldTypes.filter(_.getType == Type.INT)
+ case LongType => fieldTypes.filter(_.getType == Type.LONG)
+ case FloatType => fieldTypes.filter(_.getType == Type.FLOAT)
+ case DoubleType => fieldTypes.filter(_.getType == Type.DOUBLE)
+ case d: DecimalType => fieldTypes.filter(_.getType == Type.STRING)
+ case StringType => fieldTypes
+ .filter(x => x.getType == Type.STRING || x.getType == Type.ENUM)
+ case DateType => fieldTypes.filter(x => x.getType == Type.INT || x.getType == Type.LONG)
--- End diff --
Why long? Shouldn't we just use int for `DateType`?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by dbtsai <gi...@git.apache.org>.
Github user dbtsai commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r205684257
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -165,16 +183,112 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
result
}
- private def resolveNullableType(avroType: Schema, nullable: Boolean): Schema = {
- if (nullable) {
- // avro uses union to represent nullable type.
- val fields = avroType.getTypes.asScala
- assert(fields.length == 2)
- val actualType = fields.filter(_.getType != NULL)
- assert(actualType.length == 1)
- actualType.head
+ // Resolve an Avro union against a supplied DataType, i.e. a LongType compared against
+ // a ["null", "long"] should return a schema of type Schema.Type.LONG
+ // This function also handles resolving a DataType against unions of 2 or more types, i.e.
+ // an IntType resolves against a ["int", "long", "null"] will correctly return a schema of
+ // type Schema.Type.LONG
+ private def resolveNullableType(avroType: Schema, catalystType: DataType,
+ nullable: Boolean): Schema = {
+ (nullable, avroType.getType) match {
--- End diff --
Since the code is complicated and long, maybe it's easier to read with just old fashion `if-else`.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Merged build finished. Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by gengliangwang <gi...@git.apache.org>.
Github user gengliangwang commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r205687174
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -165,16 +183,112 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
result
}
- private def resolveNullableType(avroType: Schema, nullable: Boolean): Schema = {
- if (nullable) {
- // avro uses union to represent nullable type.
- val fields = avroType.getTypes.asScala
- assert(fields.length == 2)
- val actualType = fields.filter(_.getType != NULL)
- assert(actualType.length == 1)
- actualType.head
+ // Resolve an Avro union against a supplied DataType, i.e. a LongType compared against
+ // a ["null", "long"] should return a schema of type Schema.Type.LONG
+ // This function also handles resolving a DataType against unions of 2 or more types, i.e.
+ // an IntType resolves against a ["int", "long", "null"] will correctly return a schema of
+ // type Schema.Type.LONG
+ private def resolveNullableType(avroType: Schema, catalystType: DataType,
+ nullable: Boolean): Schema = {
+ (nullable, avroType.getType) match {
+ case (false, Type.UNION) | (true, Type.UNION) =>
+ // avro uses union to represent nullable type.
+ val fieldTypes = avroType.getTypes.asScala
+
+ // If we're nullable, we need to have at least two types. Cases with more than two types
+ // are captured in test("read read-write, read-write w/ schema, read") w/ test.avro input
+ assert(fieldTypes.length >= 2)
+
+ val actualType = catalystType match {
+ case NullType => fieldTypes.filter(_.getType == Type.NULL)
+ case BooleanType => fieldTypes.filter(_.getType == Type.BOOLEAN)
+ case ByteType => fieldTypes.filter(_.getType == Type.INT)
+ case BinaryType =>
+ val at = fieldTypes.filter(x => x.getType == Type.BYTES || x.getType == Type.FIXED)
+ if (at.length > 1) {
+ throw new IncompatibleSchemaException(
+ s"Cannot resolve schema of ${catalystType} against union ${avroType.toString}")
+ } else {
+ at
+ }
+ case ShortType | IntegerType => fieldTypes.filter(_.getType == Type.INT)
+ case LongType => fieldTypes.filter(_.getType == Type.LONG)
+ case FloatType => fieldTypes.filter(_.getType == Type.FLOAT)
+ case DoubleType => fieldTypes.filter(_.getType == Type.DOUBLE)
+ case d: DecimalType => fieldTypes.filter(_.getType == Type.STRING)
+ case StringType => fieldTypes
+ .filter(x => x.getType == Type.STRING || x.getType == Type.ENUM)
+ case DateType => fieldTypes.filter(x => x.getType == Type.INT || x.getType == Type.LONG)
+ case TimestampType => fieldTypes.filter(_.getType == Type.LONG)
+ case ArrayType(et, containsNull) =>
+ // Find array that matches the element type specified
+ fieldTypes.filter(x => x.getType == Type.ARRAY
+ && typeMatchesSchema(et, x.getElementType))
+ case st: StructType => // Find the matching record!
+ val recordTypes = fieldTypes.filter(x => x.getType == Type.RECORD)
+ if (recordTypes.length > 1) {
+ throw new IncompatibleSchemaException(
+ "Unions of multiple record types are NOT supported with user-specified schema")
+ }
+ recordTypes
+ case MapType(kt, vt, valueContainsNull) =>
+ // Find the map that matches the value type. Maps in Avro are always key type string
+ fieldTypes.filter(x => x.getType == Type.MAP && typeMatchesSchema(vt, x.getValueType))
+ case other =>
+ throw new IncompatibleSchemaException(s"Unexpected type: $other")
+ }
+
+ assert(actualType.length == 1)
--- End diff --
We need to show error message if the length is not 1.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by dbtsai <gi...@git.apache.org>.
Github user dbtsai commented on the issue:
https://github.com/apache/spark/pull/21847
@cloud-fan @gatorsmile @gengliangwang We need this to write avro files from Spark for other applications to consume, and we use the `fixed` and `enum` types in our external applications. Thanks.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #94522 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/94522/testReport)** for PR 21847 at commit [`03e112c`](https://github.com/apache/spark/commit/03e112c10c7a5f5b96c192f1388a96c74ff162f4).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #93837 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93837/testReport)** for PR 21847 at commit [`9bec5fa`](https://github.com/apache/spark/commit/9bec5fa45c8af270ab22196c4b5b3ba071106d0b).
* This patch **fails Scala style tests**.
* This patch merges cleanly.
* This patch adds the following public classes _(experimental)_:
* `trait BarrierTaskContext extends TaskContext `
* `class BarrierTaskInfo(val address: String)`
* `class RDDBarrier[T: ClassTag](rdd: RDD[T]) `
* `case class WorkerOffer(`
* `case class Shuffle(child: Expression, randomSeed: Option[Long] = None)`
* `case class ReplicateRows(children: Seq[Expression]) extends Generator with CodegenFallback `
* `trait AnalysisHelper extends QueryPlan[LogicalPlan] `
* `case class Intersect(`
* `case class Except(`
* `case class RandomIndicesGenerator(randomSeed: Long) `
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by lindblombr <gi...@git.apache.org>.
Github user lindblombr commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r206691928
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -120,7 +133,7 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
case MapType(kt, vt, valueContainsNull) if kt == StringType =>
val valueConverter = newConverter(
- vt, resolveNullableType(avroType.getValueType, valueContainsNull))
+ vt, resolveUnionType(avroType.getValueType, vt, valueContainsNull))
--- End diff --
Done.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/93858/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL][WIP]: Built-in AVRO support...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #93464 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93464/testReport)** for PR 21847 at commit [`12b3859`](https://github.com/apache/spark/commit/12b3859467155e2e205afffcc7f4db9def282bab).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #93728 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93728/testReport)** for PR 21847 at commit [`183b684`](https://github.com/apache/spark/commit/183b6841a8a121bb07d9fe553ca213eda78eb35a).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by MaxGekk <gi...@git.apache.org>.
Github user MaxGekk commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r205554970
--- Diff: external/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala ---
@@ -40,12 +41,40 @@ import org.apache.spark.sql.types._
class AvroSuite extends QueryTest with SharedSQLContext with SQLTestUtils {
val episodesAvro = testFile("episodes.avro")
val testAvro = testFile("test.avro")
+ val messyAvro = testFile("messy.avro")
+ val multiRecordTypeUnionAvro = testFile("multirecordtypeunion.avro")
+ val episodesSchemaFile = testFile("episodes.avsc")
+ val testSchemaFile = testFile("test.avsc")
+ val messySchemaFile = testFile("messy.avsc")
+ val multiRecordTypeUnionSchemaFile = testFile("multirecordtypeunion.avsc")
override protected def beforeAll(): Unit = {
super.beforeAll()
spark.conf.set("spark.sql.files.maxPartitionBytes", 1024)
}
+ def forceSchemaCheck(spark: SparkSession, inputPath: String, schemaFile: String): Unit = {
--- End diff --
`spark` is already available from `SharedSQLContext`. You don't need to pass it to the function
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by gengliangwang <gi...@git.apache.org>.
Github user gengliangwang commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r205685498
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -165,16 +183,112 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
result
}
- private def resolveNullableType(avroType: Schema, nullable: Boolean): Schema = {
- if (nullable) {
- // avro uses union to represent nullable type.
- val fields = avroType.getTypes.asScala
- assert(fields.length == 2)
- val actualType = fields.filter(_.getType != NULL)
- assert(actualType.length == 1)
- actualType.head
+ // Resolve an Avro union against a supplied DataType, i.e. a LongType compared against
+ // a ["null", "long"] should return a schema of type Schema.Type.LONG
+ // This function also handles resolving a DataType against unions of 2 or more types, i.e.
+ // an IntType resolves against a ["int", "long", "null"] will correctly return a schema of
+ // type Schema.Type.LONG
+ private def resolveNullableType(avroType: Schema, catalystType: DataType,
+ nullable: Boolean): Schema = {
+ (nullable, avroType.getType) match {
+ case (false, Type.UNION) | (true, Type.UNION) =>
+ // avro uses union to represent nullable type.
+ val fieldTypes = avroType.getTypes.asScala
+
+ // If we're nullable, we need to have at least two types. Cases with more than two types
+ // are captured in test("read read-write, read-write w/ schema, read") w/ test.avro input
+ assert(fieldTypes.length >= 2)
+
+ val actualType = catalystType match {
+ case NullType => fieldTypes.filter(_.getType == Type.NULL)
+ case BooleanType => fieldTypes.filter(_.getType == Type.BOOLEAN)
+ case ByteType => fieldTypes.filter(_.getType == Type.INT)
+ case BinaryType =>
+ val at = fieldTypes.filter(x => x.getType == Type.BYTES || x.getType == Type.FIXED)
+ if (at.length > 1) {
+ throw new IncompatibleSchemaException(
+ s"Cannot resolve schema of ${catalystType} against union ${avroType.toString}")
+ } else {
+ at
+ }
+ case ShortType | IntegerType => fieldTypes.filter(_.getType == Type.INT)
+ case LongType => fieldTypes.filter(_.getType == Type.LONG)
+ case FloatType => fieldTypes.filter(_.getType == Type.FLOAT)
+ case DoubleType => fieldTypes.filter(_.getType == Type.DOUBLE)
+ case d: DecimalType => fieldTypes.filter(_.getType == Type.STRING)
+ case StringType => fieldTypes
+ .filter(x => x.getType == Type.STRING || x.getType == Type.ENUM)
+ case DateType => fieldTypes.filter(x => x.getType == Type.INT || x.getType == Type.LONG)
+ case TimestampType => fieldTypes.filter(_.getType == Type.LONG)
+ case ArrayType(et, containsNull) =>
+ // Find array that matches the element type specified
+ fieldTypes.filter(x => x.getType == Type.ARRAY
+ && typeMatchesSchema(et, x.getElementType))
+ case st: StructType => // Find the matching record!
+ val recordTypes = fieldTypes.filter(x => x.getType == Type.RECORD)
+ if (recordTypes.length > 1) {
+ throw new IncompatibleSchemaException(
+ "Unions of multiple record types are NOT supported with user-specified schema")
+ }
+ recordTypes
+ case MapType(kt, vt, valueContainsNull) =>
+ // Find the map that matches the value type. Maps in Avro are always key type string
+ fieldTypes.filter(x => x.getType == Type.MAP && typeMatchesSchema(vt, x.getValueType))
+ case other =>
+ throw new IncompatibleSchemaException(s"Unexpected type: $other")
+ }
+
+ assert(actualType.length == 1)
+ actualType.head
+ case (false, _) | (true, _) => avroType
--- End diff --
case _ => avroType
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL][WIP]: Built-in AVRO support...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Can one of the admins verify this patch?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL][WIP]: Built-in AVRO support...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r206623930
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -165,16 +182,118 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
result
}
- private def resolveNullableType(avroType: Schema, nullable: Boolean): Schema = {
- if (nullable) {
+ // Resolve an Avro union against a supplied DataType, i.e. a LongType compared against
+ // a ["null", "long"] should return a schema of type Schema.Type.LONG
+ // This function also handles resolving a DataType against unions of 2 or more types, i.e.
+ // an IntType resolves against a ["int", "long", "null"] will correctly return a schema of
+ // type Schema.Type.LONG
+ private def resolveUnionType(avroType: Schema, catalystType: DataType,
+ nullable: Boolean): Schema = {
+ if (avroType.getType == Type.UNION) {
// avro uses union to represent nullable type.
- val fields = avroType.getTypes.asScala
- assert(fields.length == 2)
- val actualType = fields.filter(_.getType != NULL)
- assert(actualType.length == 1)
+ val fieldTypes = avroType.getTypes.asScala
+
+ // If we're nullable, we need to have at least two types. Cases with more than two types
+ // are captured in test("read read-write, read-write w/ schema, read") w/ test.avro input
+ if (nullable && fieldTypes.length < 2) {
+ throw new IncompatibleSchemaException(
+ s"Cannot resolve nullable ${catalystType} against union type ${avroType}")
+ }
+
+ val actualType = catalystType match {
+ case NullType => fieldTypes.filter(_.getType == Type.NULL)
+ case BooleanType => fieldTypes.filter(_.getType == Type.BOOLEAN)
+ case ByteType => fieldTypes.filter(_.getType == Type.INT)
+ case BinaryType =>
+ val at = fieldTypes.filter(x => x.getType == Type.BYTES || x.getType == Type.FIXED)
+ if (at.length > 1) {
+ throw new IncompatibleSchemaException(
+ s"Cannot resolve schema of ${catalystType} against union ${avroType.toString}")
+ } else {
+ at
+ }
+ case ShortType | IntegerType => fieldTypes.filter(_.getType == Type.INT)
+ case LongType => fieldTypes.filter(_.getType == Type.LONG)
+ case FloatType => fieldTypes.filter(_.getType == Type.FLOAT)
+ case DoubleType => fieldTypes.filter(_.getType == Type.DOUBLE)
+ case d: DecimalType => fieldTypes.filter(_.getType == Type.STRING)
+ case StringType => fieldTypes
+ .filter(x => x.getType == Type.STRING || x.getType == Type.ENUM)
+ case DateType => fieldTypes.filter(x => x.getType == Type.INT || x.getType == Type.LONG)
+ case TimestampType => fieldTypes.filter(_.getType == Type.LONG)
+ case ArrayType(et, containsNull) =>
+ // Find array that matches the element type specified
+ fieldTypes.filter(x => x.getType == Type.ARRAY
+ && typeMatchesSchema(et, x.getElementType))
+ case st: StructType => // Find the matching record!
+ val recordTypes = fieldTypes.filter(x => x.getType == Type.RECORD)
+ if (recordTypes.length > 1) {
+ throw new IncompatibleSchemaException(
+ "Unions of multiple record types are NOT supported with user-specified schema")
+ }
+ recordTypes
+ case MapType(kt, vt, valueContainsNull) =>
+ // Find the map that matches the value type. Maps in Avro are always key type string
+ fieldTypes.filter(x => x.getType == Type.MAP && typeMatchesSchema(vt, x.getValueType))
--- End diff --
I see here it said `Maps in Avro are always key type string`, so I'm wondering if `kt` should be `StringType` or not?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by lindblombr <gi...@git.apache.org>.
Github user lindblombr commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r205912461
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -165,16 +183,112 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
result
}
- private def resolveNullableType(avroType: Schema, nullable: Boolean): Schema = {
- if (nullable) {
- // avro uses union to represent nullable type.
- val fields = avroType.getTypes.asScala
- assert(fields.length == 2)
- val actualType = fields.filter(_.getType != NULL)
- assert(actualType.length == 1)
- actualType.head
+ // Resolve an Avro union against a supplied DataType, i.e. a LongType compared against
+ // a ["null", "long"] should return a schema of type Schema.Type.LONG
+ // This function also handles resolving a DataType against unions of 2 or more types, i.e.
+ // an IntType resolves against a ["int", "long", "null"] will correctly return a schema of
+ // type Schema.Type.LONG
+ private def resolveNullableType(avroType: Schema, catalystType: DataType,
+ nullable: Boolean): Schema = {
+ (nullable, avroType.getType) match {
+ case (false, Type.UNION) | (true, Type.UNION) =>
+ // avro uses union to represent nullable type.
+ val fieldTypes = avroType.getTypes.asScala
+
+ // If we're nullable, we need to have at least two types. Cases with more than two types
+ // are captured in test("read read-write, read-write w/ schema, read") w/ test.avro input
+ assert(fieldTypes.length >= 2)
--- End diff --
I cleaned up this logic to be more clear.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by dbtsai <gi...@git.apache.org>.
Github user dbtsai commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r209031420
--- Diff: external/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala ---
@@ -725,6 +744,158 @@ class AvroSuite extends QueryTest with SharedSQLContext with SQLTestUtils {
assert(result === Row("foo"))
}
+ test("support user provided avro schema for writing nullable enum type") {
+ withTempPath { tempDir =>
+ val avroSchema =
+ """
+ |{
+ | "type" : "record",
+ | "name" : "test_schema",
+ | "fields" : [{
+ | "name": "enum",
+ | "type": [{ "type": "enum",
+ | "name": "Suit",
+ | "symbols" : ["SPADES", "HEARTS", "DIAMONDS", "CLUBS"]
+ | }, "null"]
+ | }]
+ |}
+ """.stripMargin
+
+ val df = spark.createDataFrame(spark.sparkContext.parallelize(Seq(
+ Row("SPADES"), Row(null), Row("HEARTS"), Row("DIAMONDS"),
+ Row(null), Row("CLUBS"), Row("HEARTS"), Row("SPADES"))),
+ StructType(Seq(StructField("Suit", StringType, true))))
+
+ val tempSaveDir = s"$tempDir/save/"
+
+ df.write.format("avro").option("avroSchema", avroSchema).save(tempSaveDir)
+
+ checkAnswer(df, spark.read.format("avro").load(tempSaveDir))
+ checkAvroSchemaEquals(avroSchema, getAvroSchemaStringFromFiles(tempSaveDir))
+
+ // Writing df containing data not in the enum will throw an exception
+ intercept[SparkException] {
--- End diff --
I was thinking to do a followup PR on this. What's happening is that avro builds a map of key to indices, and avro looks up the table to get the index. In avro, they don't do any check, so the exception is null pointer exception when looking up non-existent key.
Since I think checking a message against null pointer exception is fragile, I decide to leave it to a followup PR.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by dbtsai <gi...@git.apache.org>.
Github user dbtsai commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r209116845
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -87,10 +87,36 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
(getter, ordinal) => getter.getDouble(ordinal)
case d: DecimalType =>
(getter, ordinal) => getter.getDecimal(ordinal, d.precision, d.scale).toString
- case StringType =>
- (getter, ordinal) => new Utf8(getter.getUTF8String(ordinal).getBytes)
- case BinaryType =>
- (getter, ordinal) => ByteBuffer.wrap(getter.getBinary(ordinal))
+ case StringType => avroType.getType match {
+ case Type.ENUM =>
+ import scala.collection.JavaConverters._
+ val enumSymbols: Set[String] = avroType.getEnumSymbols.asScala.toSet
+ (getter, ordinal) =>
+ val data = getter.getUTF8String(ordinal).toString
+ if (!enumSymbols.contains(data)) {
+ throw new IncompatibleSchemaException(
+ "Cannot write \"" + data + "\" since it's not defined in enum \"" +
+ enumSymbols.mkString("\", \""))
--- End diff --
Good catch.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/94478/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by lindblombr <gi...@git.apache.org>.
Github user lindblombr commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r206607145
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -165,16 +182,118 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
result
}
- private def resolveNullableType(avroType: Schema, nullable: Boolean): Schema = {
- if (nullable) {
+ // Resolve an Avro union against a supplied DataType, i.e. a LongType compared against
+ // a ["null", "long"] should return a schema of type Schema.Type.LONG
+ // This function also handles resolving a DataType against unions of 2 or more types, i.e.
+ // an IntType resolves against a ["int", "long", "null"] will correctly return a schema of
+ // type Schema.Type.LONG
+ private def resolveUnionType(avroType: Schema, catalystType: DataType,
+ nullable: Boolean): Schema = {
+ if (avroType.getType == Type.UNION) {
// avro uses union to represent nullable type.
- val fields = avroType.getTypes.asScala
- assert(fields.length == 2)
- val actualType = fields.filter(_.getType != NULL)
- assert(actualType.length == 1)
+ val fieldTypes = avroType.getTypes.asScala
+
+ // If we're nullable, we need to have at least two types. Cases with more than two types
+ // are captured in test("read read-write, read-write w/ schema, read") w/ test.avro input
+ if (nullable && fieldTypes.length < 2) {
+ throw new IncompatibleSchemaException(
+ s"Cannot resolve nullable ${catalystType} against union type ${avroType}")
+ }
+
+ val actualType = catalystType match {
+ case NullType => fieldTypes.filter(_.getType == Type.NULL)
+ case BooleanType => fieldTypes.filter(_.getType == Type.BOOLEAN)
+ case ByteType => fieldTypes.filter(_.getType == Type.INT)
+ case BinaryType =>
+ val at = fieldTypes.filter(x => x.getType == Type.BYTES || x.getType == Type.FIXED)
+ if (at.length > 1) {
+ throw new IncompatibleSchemaException(
+ s"Cannot resolve schema of ${catalystType} against union ${avroType.toString}")
+ } else {
+ at
+ }
+ case ShortType | IntegerType => fieldTypes.filter(_.getType == Type.INT)
+ case LongType => fieldTypes.filter(_.getType == Type.LONG)
+ case FloatType => fieldTypes.filter(_.getType == Type.FLOAT)
+ case DoubleType => fieldTypes.filter(_.getType == Type.DOUBLE)
+ case d: DecimalType => fieldTypes.filter(_.getType == Type.STRING)
+ case StringType => fieldTypes
+ .filter(x => x.getType == Type.STRING || x.getType == Type.ENUM)
+ case DateType => fieldTypes.filter(x => x.getType == Type.INT || x.getType == Type.LONG)
+ case TimestampType => fieldTypes.filter(_.getType == Type.LONG)
+ case ArrayType(et, containsNull) =>
+ // Find array that matches the element type specified
+ fieldTypes.filter(x => x.getType == Type.ARRAY
+ && typeMatchesSchema(et, x.getElementType))
+ case st: StructType => // Find the matching record!
+ val recordTypes = fieldTypes.filter(x => x.getType == Type.RECORD)
+ if (recordTypes.length > 1) {
+ throw new IncompatibleSchemaException(
+ "Unions of multiple record types are NOT supported with user-specified schema")
+ }
+ recordTypes
+ case MapType(kt, vt, valueContainsNull) =>
+ // Find the map that matches the value type. Maps in Avro are always key type string
+ fieldTypes.filter(x => x.getType == Type.MAP && typeMatchesSchema(vt, x.getValueType))
--- End diff --
Is the suggestion to add an explicit check for kt to be a String? Sorry if I misunderstand.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #94460 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/94460/testReport)** for PR 21847 at commit [`46fec45`](https://github.com/apache/spark/commit/46fec4519267ef28824c7745fa30f1c1f6a7b056).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #93857 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93857/testReport)** for PR 21847 at commit [`13ce570`](https://github.com/apache/spark/commit/13ce5704fd1c9df9a019e960b77494dce5eb8d83).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL][WIP]: Built-in AVRO support...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Merged build finished. Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #94190 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/94190/testReport)** for PR 21847 at commit [`02a86b9`](https://github.com/apache/spark/commit/02a86b9703ce5468c717550ea4e2a24f0f3cdbee).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL][WIP]: Built-in AVRO support...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #93453 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93453/testReport)** for PR 21847 at commit [`30fc1ae`](https://github.com/apache/spark/commit/30fc1ae98a2c29824b83e956c2cbe87e9bbb3749).
* This patch **fails Scala style tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #94190 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/94190/testReport)** for PR 21847 at commit [`02a86b9`](https://github.com/apache/spark/commit/02a86b9703ce5468c717550ea4e2a24f0f3cdbee).
* This patch **fails Scala style tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/93699/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by dbtsai <gi...@git.apache.org>.
Github user dbtsai commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r205685728
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -165,16 +183,112 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
result
}
- private def resolveNullableType(avroType: Schema, nullable: Boolean): Schema = {
- if (nullable) {
- // avro uses union to represent nullable type.
- val fields = avroType.getTypes.asScala
- assert(fields.length == 2)
- val actualType = fields.filter(_.getType != NULL)
- assert(actualType.length == 1)
- actualType.head
+ // Resolve an Avro union against a supplied DataType, i.e. a LongType compared against
+ // a ["null", "long"] should return a schema of type Schema.Type.LONG
+ // This function also handles resolving a DataType against unions of 2 or more types, i.e.
+ // an IntType resolves against a ["int", "long", "null"] will correctly return a schema of
+ // type Schema.Type.LONG
+ private def resolveNullableType(avroType: Schema, catalystType: DataType,
+ nullable: Boolean): Schema = {
+ (nullable, avroType.getType) match {
+ case (false, Type.UNION) | (true, Type.UNION) =>
+ // avro uses union to represent nullable type.
+ val fieldTypes = avroType.getTypes.asScala
+
+ // If we're nullable, we need to have at least two types. Cases with more than two types
+ // are captured in test("read read-write, read-write w/ schema, read") w/ test.avro input
+ assert(fieldTypes.length >= 2)
--- End diff --
When it's non-nullable, is it possible to have `fieldTypes.length == 1`?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/94184/
Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by dbtsai <gi...@git.apache.org>.
Github user dbtsai commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r206350423
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -87,17 +87,30 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
case d: DecimalType =>
(getter, ordinal) => getter.getDecimal(ordinal, d.precision, d.scale).toString
case StringType =>
- (getter, ordinal) => new Utf8(getter.getUTF8String(ordinal).getBytes)
+ if (avroType.getType == Type.ENUM) {
+ (getter, ordinal) =>
+ new EnumSymbol(avroType, getter.getUTF8String(ordinal).toString)
+ } else {
+ (getter, ordinal) =>
+ new Utf8(getter.getUTF8String(ordinal).getBytes)
+ }
case BinaryType =>
- (getter, ordinal) => ByteBuffer.wrap(getter.getBinary(ordinal))
+ if (avroType.getType == Type.FIXED) {
+ // Handles fixed-type fields in output schema. Test case is included in test.avro
+ // as it includes several fixed fields that would fail if we specify schema
+ // on-write without this condition
+ (getter, ordinal) => new Fixed(avroType, getter.getBinary(ordinal))
+ } else {
+ (getter, ordinal) => ByteBuffer.wrap(getter.getBinary(ordinal))
+ }
case DateType =>
(getter, ordinal) => getter.getInt(ordinal) * DateTimeUtils.MILLIS_PER_DAY
case TimestampType =>
(getter, ordinal) => getter.getLong(ordinal) / 1000
case ArrayType(et, containsNull) =>
val elementConverter = newConverter(
- et, resolveNullableType(avroType.getElementType, containsNull))
+ et, resolveUnionType(avroType.getElementType, et, containsNull))
(getter, ordinal) => {
--- End diff --
With `if (containsNull && arrayData.isNullAt(i))`, JVM can remove the branching if it's not nullable. We can use array instead of arraybuffer to get better performance.
```scala
(getter, ordinal) => {
val arrayData = getter.getArray(ordinal)
val length = arrayData.numElements()
val result = new Array[Any](length)
var i = 0
while (i < length) {
if (containsNull && arrayData.isNullAt(i)) {
result(i) = null
} else {
result(i) = elementConverter(arrayData, i)
}
i += 1
}
result
}
```
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL][WIP]: Built-in AVRO support...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/93604/
Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r206369093
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -165,16 +182,118 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
result
}
- private def resolveNullableType(avroType: Schema, nullable: Boolean): Schema = {
- if (nullable) {
+ // Resolve an Avro union against a supplied DataType, i.e. a LongType compared against
+ // a ["null", "long"] should return a schema of type Schema.Type.LONG
+ // This function also handles resolving a DataType against unions of 2 or more types, i.e.
+ // an IntType resolves against a ["int", "long", "null"] will correctly return a schema of
+ // type Schema.Type.LONG
+ private def resolveUnionType(avroType: Schema, catalystType: DataType,
+ nullable: Boolean): Schema = {
+ if (avroType.getType == Type.UNION) {
// avro uses union to represent nullable type.
- val fields = avroType.getTypes.asScala
- assert(fields.length == 2)
- val actualType = fields.filter(_.getType != NULL)
- assert(actualType.length == 1)
+ val fieldTypes = avroType.getTypes.asScala
+
+ // If we're nullable, we need to have at least two types. Cases with more than two types
+ // are captured in test("read read-write, read-write w/ schema, read") w/ test.avro input
+ if (nullable && fieldTypes.length < 2) {
+ throw new IncompatibleSchemaException(
+ s"Cannot resolve nullable ${catalystType} against union type ${avroType}")
+ }
+
+ val actualType = catalystType match {
+ case NullType => fieldTypes.filter(_.getType == Type.NULL)
+ case BooleanType => fieldTypes.filter(_.getType == Type.BOOLEAN)
+ case ByteType => fieldTypes.filter(_.getType == Type.INT)
+ case BinaryType =>
+ val at = fieldTypes.filter(x => x.getType == Type.BYTES || x.getType == Type.FIXED)
+ if (at.length > 1) {
--- End diff --
Do we need this check here? Seems L242 check will also handle the case?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/93621/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL][WIP]: Built-in AVRO support...
Posted by dbtsai <gi...@git.apache.org>.
Github user dbtsai commented on the issue:
https://github.com/apache/spark/pull/21847
add to whitelist
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #94524 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/94524/testReport)** for PR 21847 at commit [`6805c58`](https://github.com/apache/spark/commit/6805c58d0c6fc30addbe2d790a08c29532921f44).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #94184 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/94184/testReport)** for PR 21847 at commit [`5cf20df`](https://github.com/apache/spark/commit/5cf20dfa89a74503ad9befed2c7e0c791aea7763).
* This patch **fails Scala style tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by dbtsai <gi...@git.apache.org>.
Github user dbtsai commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r205692946
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -165,16 +183,112 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
result
}
- private def resolveNullableType(avroType: Schema, nullable: Boolean): Schema = {
- if (nullable) {
- // avro uses union to represent nullable type.
- val fields = avroType.getTypes.asScala
- assert(fields.length == 2)
- val actualType = fields.filter(_.getType != NULL)
- assert(actualType.length == 1)
- actualType.head
+ // Resolve an Avro union against a supplied DataType, i.e. a LongType compared against
+ // a ["null", "long"] should return a schema of type Schema.Type.LONG
+ // This function also handles resolving a DataType against unions of 2 or more types, i.e.
+ // an IntType resolves against a ["int", "long", "null"] will correctly return a schema of
+ // type Schema.Type.LONG
+ private def resolveNullableType(avroType: Schema, catalystType: DataType,
+ nullable: Boolean): Schema = {
+ (nullable, avroType.getType) match {
+ case (false, Type.UNION) | (true, Type.UNION) =>
+ // avro uses union to represent nullable type.
+ val fieldTypes = avroType.getTypes.asScala
+
+ // If we're nullable, we need to have at least two types. Cases with more than two types
+ // are captured in test("read read-write, read-write w/ schema, read") w/ test.avro input
+ assert(fieldTypes.length >= 2)
+
+ val actualType = catalystType match {
+ case NullType => fieldTypes.filter(_.getType == Type.NULL)
+ case BooleanType => fieldTypes.filter(_.getType == Type.BOOLEAN)
+ case ByteType => fieldTypes.filter(_.getType == Type.INT)
+ case BinaryType =>
+ val at = fieldTypes.filter(x => x.getType == Type.BYTES || x.getType == Type.FIXED)
+ if (at.length > 1) {
+ throw new IncompatibleSchemaException(
+ s"Cannot resolve schema of ${catalystType} against union ${avroType.toString}")
+ } else {
+ at
+ }
+ case ShortType | IntegerType => fieldTypes.filter(_.getType == Type.INT)
+ case LongType => fieldTypes.filter(_.getType == Type.LONG)
+ case FloatType => fieldTypes.filter(_.getType == Type.FLOAT)
+ case DoubleType => fieldTypes.filter(_.getType == Type.DOUBLE)
+ case d: DecimalType => fieldTypes.filter(_.getType == Type.STRING)
+ case StringType => fieldTypes
+ .filter(x => x.getType == Type.STRING || x.getType == Type.ENUM)
+ case DateType => fieldTypes.filter(x => x.getType == Type.INT || x.getType == Type.LONG)
+ case TimestampType => fieldTypes.filter(_.getType == Type.LONG)
+ case ArrayType(et, containsNull) =>
+ // Find array that matches the element type specified
+ fieldTypes.filter(x => x.getType == Type.ARRAY
+ && typeMatchesSchema(et, x.getElementType))
+ case st: StructType => // Find the matching record!
+ val recordTypes = fieldTypes.filter(x => x.getType == Type.RECORD)
+ if (recordTypes.length > 1) {
+ throw new IncompatibleSchemaException(
+ "Unions of multiple record types are NOT supported with user-specified schema")
+ }
+ recordTypes
+ case MapType(kt, vt, valueContainsNull) =>
+ // Find the map that matches the value type. Maps in Avro are always key type string
+ fieldTypes.filter(x => x.getType == Type.MAP && typeMatchesSchema(vt, x.getValueType))
+ case other =>
+ throw new IncompatibleSchemaException(s"Unexpected type: $other")
+ }
+
+ assert(actualType.length == 1)
+ actualType.head
+ case (false, _) | (true, _) => avroType
+ }
+ }
+
+ // Given a Schema and a DataType, do they match?
+ private def typeMatchesSchema(catalystType: DataType, avroSchema: Schema): Boolean = {
+ if (catalystType.isInstanceOf[StructType]) {
+ val avroFields = resolveNullableType(avroSchema, catalystType,
+ avroSchema.getType == Type.UNION)
+ .getFields
+ if (avroFields.size() == catalystType.asInstanceOf[StructType].length) {
+ catalystType.asInstanceOf[StructType].zip(avroFields.asScala).forall {
+ case (f1, f2) => typeMatchesSchema(f1.dataType, f2.schema)
+ }
+ } else {
+ false
+ }
} else {
- avroType
+ val isTypeCompatible = (a: Schema, b: DataType, c: Type) =>
+ resolveNullableType(a, b, a.getType == Type.UNION).getType == c
+
+ catalystType match {
+ case ByteType | ShortType | IntegerType =>
+ isTypeCompatible(avroSchema, catalystType, Type.INT)
+ case BooleanType => isTypeCompatible(avroSchema, catalystType, Type.BOOLEAN)
+ case BinaryType => isTypeCompatible(avroSchema, catalystType, Type.BYTES)
+ case LongType | TimestampType => isTypeCompatible(avroSchema, catalystType, Type.LONG)
+ case FloatType => isTypeCompatible(avroSchema, catalystType, Type.FLOAT)
+ case DoubleType => isTypeCompatible(avroSchema, catalystType, Type.DOUBLE)
+ case d: DecimalType =>
+ // newConverter always returns a string representation for DecimalType, so we honor
+ // that here, since we don't yet support Avro's logical types
+ isTypeCompatible(avroSchema, catalystType, Type.STRING)
+ case StringType => isTypeCompatible(avroSchema, catalystType, Type.STRING) ||
+ isTypeCompatible(avroSchema, catalystType, Type.ENUM)
+ case DateType => isTypeCompatible(avroSchema, catalystType, Type.INT) ||
+ isTypeCompatible(avroSchema, catalystType, Type.LONG)
+ case ArrayType(et, containsNull) =>
+ isTypeCompatible(avroSchema, catalystType, Type.ARRAY) &&
+ typeMatchesSchema(et,
+ resolveNullableType(avroSchema, catalystType, avroSchema.getType == Type.UNION)
+ .getElementType)
+ case MapType(kt, vt, valueContainsNull) =>
+ isTypeCompatible(avroSchema, catalystType, Type.MAP) &&
+ typeMatchesSchema(vt,
+ resolveNullableType(avroSchema, catalystType, avroSchema.getType == Type.UNION)
+ .getValueType)
+ }
}
+
--- End diff --
remove this extra line.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Merged build finished. Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #94524 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/94524/testReport)** for PR 21847 at commit [`6805c58`](https://github.com/apache/spark/commit/6805c58d0c6fc30addbe2d790a08c29532921f44).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by gengliangwang <gi...@git.apache.org>.
Github user gengliangwang commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r205685302
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -165,16 +183,112 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
result
}
- private def resolveNullableType(avroType: Schema, nullable: Boolean): Schema = {
- if (nullable) {
- // avro uses union to represent nullable type.
- val fields = avroType.getTypes.asScala
- assert(fields.length == 2)
- val actualType = fields.filter(_.getType != NULL)
- assert(actualType.length == 1)
- actualType.head
+ // Resolve an Avro union against a supplied DataType, i.e. a LongType compared against
+ // a ["null", "long"] should return a schema of type Schema.Type.LONG
+ // This function also handles resolving a DataType against unions of 2 or more types, i.e.
+ // an IntType resolves against a ["int", "long", "null"] will correctly return a schema of
+ // type Schema.Type.LONG
+ private def resolveNullableType(avroType: Schema, catalystType: DataType,
--- End diff --
rename to `resolveUnionType`?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL][WIP]: Built-in AVRO support...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Can one of the admins verify this patch?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #93846 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93846/testReport)** for PR 21847 at commit [`241e5af`](https://github.com/apache/spark/commit/241e5af6b8591f822e2b591ae80afd9c5abaab03).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on the issue:
https://github.com/apache/spark/pull/21847
Few minor comments. LGTM.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/93626/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/93625/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #93846 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93846/testReport)** for PR 21847 at commit [`241e5af`](https://github.com/apache/spark/commit/241e5af6b8591f822e2b591ae80afd9c5abaab03).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #93858 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93858/testReport)** for PR 21847 at commit [`42978fb`](https://github.com/apache/spark/commit/42978fb13ce5d292f8c4e6a257e6ea14fc8b0670).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by MaxGekk <gi...@git.apache.org>.
Github user MaxGekk commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r205556740
--- Diff: external/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala ---
@@ -40,12 +41,40 @@ import org.apache.spark.sql.types._
class AvroSuite extends QueryTest with SharedSQLContext with SQLTestUtils {
val episodesAvro = testFile("episodes.avro")
val testAvro = testFile("test.avro")
+ val messyAvro = testFile("messy.avro")
+ val multiRecordTypeUnionAvro = testFile("multirecordtypeunion.avro")
+ val episodesSchemaFile = testFile("episodes.avsc")
+ val testSchemaFile = testFile("test.avsc")
+ val messySchemaFile = testFile("messy.avsc")
+ val multiRecordTypeUnionSchemaFile = testFile("multirecordtypeunion.avsc")
override protected def beforeAll(): Unit = {
super.beforeAll()
spark.conf.set("spark.sql.files.maxPartitionBytes", 1024)
}
+ def forceSchemaCheck(spark: SparkSession, inputPath: String, schemaFile: String): Unit = {
+ withTempPath { tempDir =>
+ val df = spark.read.format("avro").load(inputPath)
+
+ val tempSaveDir1 = s"$tempDir/test1/"
+ val tempSaveDir2 = s"$tempDir/test2/"
+
+ df.write.format("avro").save(tempSaveDir1)
+
+ val newDf = spark.read.format("avro").load(tempSaveDir1)
+ checkAnswer(df, newDf)
--- End diff --
Do you actually check schemas? `checkAnswer` does collect of `newDf`, and check content of 2 dataframes. I am asking because method name `forceSchemaCheck ` confuses slightly.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #94184 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/94184/testReport)** for PR 21847 at commit [`5cf20df`](https://github.com/apache/spark/commit/5cf20dfa89a74503ad9befed2c7e0c791aea7763).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL][WIP]: Built-in AVRO support...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #93606 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93606/testReport)** for PR 21847 at commit [`7e44ca0`](https://github.com/apache/spark/commit/7e44ca008b98a7c679f6303c5a5d0c4dd327ee33).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by MaxGekk <gi...@git.apache.org>.
Github user MaxGekk commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r205561746
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -165,16 +183,100 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
result
}
- private def resolveNullableType(avroType: Schema, nullable: Boolean): Schema = {
- if (nullable) {
+ private def resolveNullableType(avroType: Schema, catalystType: DataType,
+ nullable: Boolean): Schema = {
+ if (nullable && avroType.getType == Type.UNION) {
// avro uses union to represent nullable type.
- val fields = avroType.getTypes.asScala
- assert(fields.length == 2)
- val actualType = fields.filter(_.getType != NULL)
+ val fieldTypes = avroType.getTypes.asScala
+
+ // If we're nullable, we need to have at least two types. Cases with more than two types are
+ // captured in test("read read-write, read-write w/ schema, read") w/ test.avro input
+ assert(fieldTypes.length >= 2)
+
+ val actualType = catalystType match {
+ case NullType => fieldTypes.filter(_.getType == Type.NULL)
+ case BooleanType => fieldTypes.filter(_.getType == Type.BOOLEAN)
+ case ByteType => fieldTypes.filter(_.getType == Type.INT)
+ case BinaryType =>
+ val at = fieldTypes.filter(x => x.getType == Type.BYTES || x.getType == Type.FIXED)
+ if (at.length > 1) {
+ throw new IncompatibleSchemaException (
+ s"Cannot resolve schema of ${catalystType} against union ${avroType.toString}")
+ } else {
+ at
+ }
+ case ShortType | IntegerType => fieldTypes.filter(_.getType == Type.INT)
+ case LongType => fieldTypes.filter(_.getType == Type.LONG)
+ case FloatType => fieldTypes.filter(_.getType == Type.FLOAT)
+ case DoubleType => fieldTypes.filter(_.getType == Type.DOUBLE)
+ case d: DecimalType => fieldTypes.filter(_.getType == Type.STRING)
+ case StringType => fieldTypes
+ .filter(x => x.getType == Type.STRING || x.getType == Type.ENUM)
+ case DateType => fieldTypes.filter(x => x.getType == Type.INT || x.getType == Type.LONG)
+ case TimestampType => fieldTypes.filter(_.getType == Type.LONG)
+ case ArrayType(et, containsNull) =>
+ // Find array that matches the type
+ fieldTypes.filter(x => x.getType == Type.ARRAY && typeMatchesSchema(et, x.getElementType))
+ case st: StructType => // Find the matching record!
+ val recordTypes = fieldTypes.filter(x => x.getType == Type.RECORD)
+ if (recordTypes.length > 1) {
+ throw new IncompatibleSchemaException(
+ "Unions of multiple record types are NOT supported with user-specified schema")
+ }
+ recordTypes
+ case MapType(kt, vt, valueContainsNull) =>
+ // Find the map that matches the type
+ fieldTypes.filter(x => x.getType == Type.MAP && typeMatchesSchema(vt, x.getValueType))
+ case other =>
+ throw new IncompatibleSchemaException(s"Unexpected type: $other")
+ }
+
assert(actualType.length == 1)
actualType.head
} else {
avroType
}
}
+
+ // Given a Schema and a DataType, do they match?
+ private def typeMatchesSchema(catalystType: DataType, avroSchema: Schema): Boolean = {
+ if (catalystType.isInstanceOf[StructType]) {
+ val avroFields = resolveNullableType(avroSchema, catalystType,
+ avroSchema.getType == Type.UNION)
+ .getFields
+ assert(avroFields.size() == catalystType.asInstanceOf[StructType].length)
+ catalystType.asInstanceOf[StructType].zip(avroFields.asScala).map {
+ case (f1, f2) => typeMatchesSchema(f1.dataType, f2.schema)
+ }.foldLeft(true)(_ && _)
+ } else {
+ val isTypeCompatible = (a: Schema, b: DataType, c: Type) =>
+ resolveNullableType(a, b, a.getType == Type.UNION).getType == c
+
+ catalystType match {
+ case ByteType | ShortType | IntegerType =>
+ isTypeCompatible(avroSchema, catalystType, Type.INT)
+ case BooleanType => isTypeCompatible(avroSchema, catalystType, Type.BOOLEAN)
+ case BinaryType => isTypeCompatible(avroSchema, catalystType, Type.BYTES)
+ case LongType | TimestampType => isTypeCompatible(avroSchema, catalystType, Type.LONG)
+ case FloatType => isTypeCompatible(avroSchema, catalystType, Type.FLOAT)
+ case DoubleType => isTypeCompatible(avroSchema, catalystType, Type.DOUBLE)
+ case d: DecimalType => isTypeCompatible(avroSchema, catalystType, Type.STRING)
--- End diff --
Could you add a comment why for `DecimalType` you pass `Type.STRING`
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #93857 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93857/testReport)** for PR 21847 at commit [`13ce570`](https://github.com/apache/spark/commit/13ce5704fd1c9df9a019e960b77494dce5eb8d83).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #94166 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/94166/testReport)** for PR 21847 at commit [`eebc16d`](https://github.com/apache/spark/commit/eebc16dec56aca6ae7086906ba43f0e9896075a3).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #94531 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/94531/testReport)** for PR 21847 at commit [`df486f2`](https://github.com/apache/spark/commit/df486f23d763450800bc7d8c4f03224d6fad1df0).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL][WIP]: Built-in AVRO support...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Can one of the admins verify this patch?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #93626 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93626/testReport)** for PR 21847 at commit [`c76aadd`](https://github.com/apache/spark/commit/c76aaddb40275537dded99ab0c708a6802e10a30).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL][WIP]: Built-in AVRO support...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Merged build finished. Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL][WIP]: Built-in AVRO support...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #93466 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93466/testReport)** for PR 21847 at commit [`033f4dd`](https://github.com/apache/spark/commit/033f4dd11b068072f023312b8cd5ae52fef769bc).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by dbtsai <gi...@git.apache.org>.
Github user dbtsai commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r206358703
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -165,16 +182,118 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
result
}
- private def resolveNullableType(avroType: Schema, nullable: Boolean): Schema = {
- if (nullable) {
+ // Resolve an Avro union against a supplied DataType, i.e. a LongType compared against
+ // a ["null", "long"] should return a schema of type Schema.Type.LONG
+ // This function also handles resolving a DataType against unions of 2 or more types, i.e.
+ // an IntType resolves against a ["int", "long", "null"] will correctly return a schema of
+ // type Schema.Type.LONG
+ private def resolveUnionType(avroType: Schema, catalystType: DataType,
+ nullable: Boolean): Schema = {
+ if (avroType.getType == Type.UNION) {
// avro uses union to represent nullable type.
- val fields = avroType.getTypes.asScala
- assert(fields.length == 2)
- val actualType = fields.filter(_.getType != NULL)
- assert(actualType.length == 1)
+ val fieldTypes = avroType.getTypes.asScala
+
+ // If we're nullable, we need to have at least two types. Cases with more than two types
+ // are captured in test("read read-write, read-write w/ schema, read") w/ test.avro input
+ if (nullable && fieldTypes.length < 2) {
+ throw new IncompatibleSchemaException(
+ s"Cannot resolve nullable ${catalystType} against union type ${avroType}")
+ }
+
+ val actualType = catalystType match {
+ case NullType => fieldTypes.filter(_.getType == Type.NULL)
+ case BooleanType => fieldTypes.filter(_.getType == Type.BOOLEAN)
+ case ByteType => fieldTypes.filter(_.getType == Type.INT)
+ case BinaryType =>
+ val at = fieldTypes.filter(x => x.getType == Type.BYTES || x.getType == Type.FIXED)
+ if (at.length > 1) {
+ throw new IncompatibleSchemaException(
+ s"Cannot resolve schema of ${catalystType} against union ${avroType.toString}")
+ } else {
+ at
+ }
+ case ShortType | IntegerType => fieldTypes.filter(_.getType == Type.INT)
+ case LongType => fieldTypes.filter(_.getType == Type.LONG)
+ case FloatType => fieldTypes.filter(_.getType == Type.FLOAT)
+ case DoubleType => fieldTypes.filter(_.getType == Type.DOUBLE)
+ case d: DecimalType => fieldTypes.filter(_.getType == Type.STRING)
+ case StringType => fieldTypes
+ .filter(x => x.getType == Type.STRING || x.getType == Type.ENUM)
+ case DateType => fieldTypes.filter(x => x.getType == Type.INT || x.getType == Type.LONG)
+ case TimestampType => fieldTypes.filter(_.getType == Type.LONG)
+ case ArrayType(et, containsNull) =>
+ // Find array that matches the element type specified
+ fieldTypes.filter(x => x.getType == Type.ARRAY
+ && typeMatchesSchema(et, x.getElementType))
+ case st: StructType => // Find the matching record!
+ val recordTypes = fieldTypes.filter(x => x.getType == Type.RECORD)
+ if (recordTypes.length > 1) {
+ throw new IncompatibleSchemaException(
+ "Unions of multiple record types are NOT supported with user-specified schema")
+ }
+ recordTypes
+ case MapType(kt, vt, valueContainsNull) =>
+ // Find the map that matches the value type. Maps in Avro are always key type string
+ fieldTypes.filter(x => x.getType == Type.MAP && typeMatchesSchema(vt, x.getValueType))
+ case other =>
+ throw new IncompatibleSchemaException(s"Unexpected type: $other")
+ }
+
+ if (actualType.length != 1) {
+ throw new IncompatibleSchemaException(
+ s"Failed to resolve ${catalystType} against ambiguous schema ${avroType}")
+ }
actualType.head
} else {
avroType
}
}
+
+ // Given a Schema and a DataType, do they match?
+ private def typeMatchesSchema(catalystType: DataType, avroSchema: Schema): Boolean = {
+ if (catalystType.isInstanceOf[StructType]) {
+ val avroFields = resolveUnionType(avroSchema, catalystType,
+ avroSchema.getType == Type.UNION)
+ .getFields
+ if (avroFields.size() == catalystType.asInstanceOf[StructType].length) {
+ catalystType.asInstanceOf[StructType].zip(avroFields.asScala).forall {
+ case (f1, f2) => typeMatchesSchema(f1.dataType, f2.schema)
+ }
+ } else {
+ false
+ }
+ } else {
+ val isTypeCompatible = (a: Schema, b: DataType, c: Type) =>
+ resolveUnionType(a, b, a.getType == Type.UNION).getType == c
+
+ catalystType match {
+ case ByteType | ShortType | IntegerType =>
+ isTypeCompatible(avroSchema, catalystType, Type.INT)
+ case BooleanType => isTypeCompatible(avroSchema, catalystType, Type.BOOLEAN)
+ case BinaryType => isTypeCompatible(avroSchema, catalystType, Type.BYTES)
+ case LongType | TimestampType => isTypeCompatible(avroSchema, catalystType, Type.LONG)
+ case FloatType => isTypeCompatible(avroSchema, catalystType, Type.FLOAT)
+ case DoubleType => isTypeCompatible(avroSchema, catalystType, Type.DOUBLE)
+ case d: DecimalType =>
+ // newConverter always returns a string representation for DecimalType, so we honor
+ // that here, since we don't yet support Avro's logical types
+ isTypeCompatible(avroSchema, catalystType, Type.STRING)
+ case StringType => isTypeCompatible(avroSchema, catalystType, Type.STRING) ||
+ isTypeCompatible(avroSchema, catalystType, Type.ENUM)
+ case DateType => isTypeCompatible(avroSchema, catalystType, Type.INT) ||
+ isTypeCompatible(avroSchema, catalystType, Type.LONG)
+ case ArrayType(et, containsNull) =>
+ isTypeCompatible(avroSchema, catalystType, Type.ARRAY) &&
+ typeMatchesSchema(et,
+ resolveUnionType(avroSchema, catalystType, avroSchema.getType == Type.UNION)
+ .getElementType)
+ case MapType(kt, vt, valueContainsNull) =>
+ isTypeCompatible(avroSchema, catalystType, Type.MAP) &&
+ typeMatchesSchema(vt,
+ resolveUnionType(avroSchema, catalystType, avroSchema.getType == Type.UNION)
+ .getValueType)
+ }
+ }
+
--- End diff --
remove the extra line
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by lindblombr <gi...@git.apache.org>.
Github user lindblombr commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r206737762
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -165,16 +186,117 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
result
}
- private def resolveNullableType(avroType: Schema, nullable: Boolean): Schema = {
- if (nullable) {
+ // Resolve an Avro union against a supplied DataType, i.e. a LongType compared against
+ // a ["null", "long"] should return a schema of type Schema.Type.LONG
+ // This function also handles resolving a DataType against unions of 2 or more types, i.e.
+ // an IntType resolves against a ["int", "long", "null"] will correctly return a schema of
+ // type Schema.Type.LONG
+ private def resolveUnionType(avroType: Schema, catalystType: DataType,
+ nullable: Boolean): Schema = {
+ if (avroType.getType == Type.UNION) {
// avro uses union to represent nullable type.
- val fields = avroType.getTypes.asScala
- assert(fields.length == 2)
- val actualType = fields.filter(_.getType != NULL)
- assert(actualType.length == 1)
+ val fieldTypes = avroType.getTypes.asScala
+
+ // If we're nullable, we need to have at least two types. Cases with more than two types
+ // are captured in test("read read-write, read-write w/ schema, read") w/ test.avro input
+ if (nullable && fieldTypes.length < 2) {
+ throw new IncompatibleSchemaException(
+ s"Cannot resolve nullable ${catalystType} against union type ${avroType}")
+ }
+
+ val actualType = catalystType match {
+ case NullType => fieldTypes.filter(_.getType == Type.NULL)
+ case BooleanType => fieldTypes.filter(_.getType == Type.BOOLEAN)
+ case ByteType => fieldTypes.filter(_.getType == Type.INT)
+ case BinaryType => fieldTypes
+ .filter(x => x.getType == Type.BYTES || x.getType == Type.FIXED)
+ case ShortType | IntegerType => fieldTypes.filter(_.getType == Type.INT)
+ case LongType => fieldTypes.filter(_.getType == Type.LONG)
+ case FloatType => fieldTypes.filter(_.getType == Type.FLOAT)
+ case DoubleType => fieldTypes.filter(_.getType == Type.DOUBLE)
+ case d: DecimalType => fieldTypes.filter(_.getType == Type.STRING)
+ case StringType => fieldTypes
+ .filter(x => x.getType == Type.STRING || x.getType == Type.ENUM)
+ case DateType => fieldTypes
+ .filter(x => x.getType == Type.INT || x.getType == Type.LONG)
--- End diff --
`DateType` is actually `long`, according to `SchemaConverters.toAvroType` L116:
catalystType match {
...
case DateType => builder.longType()
If anything, it appears I don't have to consider `int`. I've removed it.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL][WIP]: Built-in AVRO support...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #93606 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93606/testReport)** for PR 21847 at commit [`7e44ca0`](https://github.com/apache/spark/commit/7e44ca008b98a7c679f6303c5a5d0c4dd327ee33).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by dbtsai <gi...@git.apache.org>.
Github user dbtsai commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r208818671
--- Diff: external/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala ---
@@ -78,6 +79,25 @@ class AvroSuite extends QueryTest with SharedSQLContext with SQLTestUtils {
checkAnswer(newEntries, originalEntries)
}
+ def checkAvroSchemaEquals(avroSchema: String, expectedAvroSchema: String): Unit = {
+ assert(new Schema.Parser().parse(avroSchema) ==
+ new Schema.Parser().parse(expectedAvroSchema))
+ }
+
+ def getAvroSchemaStringFromFiles(filePath: String): String = {
--- End diff --
Good suggestion. I'll address it in the next push.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r206369173
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -165,16 +182,118 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
result
}
- private def resolveNullableType(avroType: Schema, nullable: Boolean): Schema = {
- if (nullable) {
+ // Resolve an Avro union against a supplied DataType, i.e. a LongType compared against
+ // a ["null", "long"] should return a schema of type Schema.Type.LONG
+ // This function also handles resolving a DataType against unions of 2 or more types, i.e.
+ // an IntType resolves against a ["int", "long", "null"] will correctly return a schema of
+ // type Schema.Type.LONG
+ private def resolveUnionType(avroType: Schema, catalystType: DataType,
+ nullable: Boolean): Schema = {
+ if (avroType.getType == Type.UNION) {
// avro uses union to represent nullable type.
- val fields = avroType.getTypes.asScala
- assert(fields.length == 2)
- val actualType = fields.filter(_.getType != NULL)
- assert(actualType.length == 1)
+ val fieldTypes = avroType.getTypes.asScala
+
+ // If we're nullable, we need to have at least two types. Cases with more than two types
+ // are captured in test("read read-write, read-write w/ schema, read") w/ test.avro input
+ if (nullable && fieldTypes.length < 2) {
+ throw new IncompatibleSchemaException(
+ s"Cannot resolve nullable ${catalystType} against union type ${avroType}")
+ }
+
+ val actualType = catalystType match {
+ case NullType => fieldTypes.filter(_.getType == Type.NULL)
+ case BooleanType => fieldTypes.filter(_.getType == Type.BOOLEAN)
+ case ByteType => fieldTypes.filter(_.getType == Type.INT)
+ case BinaryType =>
+ val at = fieldTypes.filter(x => x.getType == Type.BYTES || x.getType == Type.FIXED)
+ if (at.length > 1) {
+ throw new IncompatibleSchemaException(
+ s"Cannot resolve schema of ${catalystType} against union ${avroType.toString}")
+ } else {
+ at
+ }
+ case ShortType | IntegerType => fieldTypes.filter(_.getType == Type.INT)
+ case LongType => fieldTypes.filter(_.getType == Type.LONG)
+ case FloatType => fieldTypes.filter(_.getType == Type.FLOAT)
+ case DoubleType => fieldTypes.filter(_.getType == Type.DOUBLE)
+ case d: DecimalType => fieldTypes.filter(_.getType == Type.STRING)
+ case StringType => fieldTypes
+ .filter(x => x.getType == Type.STRING || x.getType == Type.ENUM)
+ case DateType => fieldTypes.filter(x => x.getType == Type.INT || x.getType == Type.LONG)
+ case TimestampType => fieldTypes.filter(_.getType == Type.LONG)
+ case ArrayType(et, containsNull) =>
+ // Find array that matches the element type specified
+ fieldTypes.filter(x => x.getType == Type.ARRAY
+ && typeMatchesSchema(et, x.getElementType))
+ case st: StructType => // Find the matching record!
--- End diff --
Move the comment next line as others?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by dbtsai <gi...@git.apache.org>.
Github user dbtsai commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r209117054
--- Diff: external/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala ---
@@ -725,6 +744,205 @@ class AvroSuite extends QueryTest with SharedSQLContext with SQLTestUtils {
assert(result === Row("foo"))
}
+ test("support user provided avro schema for writing nullable enum type") {
+ withTempPath { tempDir =>
+ val avroSchema =
+ """
+ |{
+ | "type" : "record",
+ | "name" : "test_schema",
+ | "fields" : [{
+ | "name": "enum",
+ | "type": [{ "type": "enum",
+ | "name": "Suit",
+ | "symbols" : ["SPADES", "HEARTS", "DIAMONDS", "CLUBS"]
+ | }, "null"]
+ | }]
+ |}
+ """.stripMargin
+
+ val df = spark.createDataFrame(spark.sparkContext.parallelize(Seq(
+ Row("SPADES"), Row(null), Row("HEARTS"), Row("DIAMONDS"),
+ Row(null), Row("CLUBS"), Row("HEARTS"), Row("SPADES"))),
+ StructType(Seq(StructField("Suit", StringType, true))))
+
+ val tempSaveDir = s"$tempDir/save/"
+
+ df.write.format("avro").option("avroSchema", avroSchema).save(tempSaveDir)
+
+ checkAnswer(df, spark.read.format("avro").load(tempSaveDir))
+ checkAvroSchemaEquals(avroSchema, getAvroSchemaStringFromFiles(tempSaveDir))
+
+ // Writing df containing data not in the enum will throw an exception
+ val message = intercept[SparkException] {
+ spark.createDataFrame(spark.sparkContext.parallelize(Seq(
+ Row("SPADES"), Row("NOT-IN-ENUM"), Row("HEARTS"), Row("DIAMONDS"))),
+ StructType(Seq(StructField("Suit", StringType, true))))
+ .write.format("avro").option("avroSchema", avroSchema)
+ .save(s"$tempDir/${UUID.randomUUID()}")
+ }.getCause.getMessage
+ assert(message.contains("org.apache.spark.sql.avro.IncompatibleSchemaException: " +
+ "Cannot write \"NOT-IN-ENUM\" since it's not defined in enum"))
+ }
+ }
+
+ test("support user provided avro schema for writing non-nullable enum type") {
+ withTempPath { tempDir =>
+ val avroSchema =
+ """
+ |{
+ | "type" : "record",
+ | "name" : "test_schema",
+ | "fields" : [{
+ | "name": "enum",
+ | "type": { "type": "enum",
+ | "name": "Suit",
+ | "symbols" : ["SPADES", "HEARTS", "DIAMONDS", "CLUBS"]
+ | }
+ | }]
+ |}
+ """.stripMargin
+
+ val dfWithNull = spark.createDataFrame(spark.sparkContext.parallelize(Seq(
+ Row("SPADES"), Row(null), Row("HEARTS"), Row("DIAMONDS"),
+ Row(null), Row("CLUBS"), Row("HEARTS"), Row("SPADES"))),
+ StructType(Seq(StructField("Suit", StringType, true))))
+
+ val df = spark.createDataFrame(dfWithNull.na.drop().rdd,
+ StructType(Seq(StructField("Suit", StringType, false))))
+
+ val tempSaveDir = s"$tempDir/save1/"
--- End diff --
addressed
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/94523/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #93619 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93619/testReport)** for PR 21847 at commit [`d85242e`](https://github.com/apache/spark/commit/d85242e1dd1b24847dc1fca8c8040cb72e9fab53).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Merged build finished. Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #93603 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93603/testReport)** for PR 21847 at commit [`be037bd`](https://github.com/apache/spark/commit/be037bda9dbb1eee267ccf5ff1391d52cae93714).
* This patch **fails Spark unit tests**.
* This patch **does not merge cleanly**.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #93728 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93728/testReport)** for PR 21847 at commit [`183b684`](https://github.com/apache/spark/commit/183b6841a8a121bb07d9fe553ca213eda78eb35a).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r209115631
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -87,10 +87,36 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
(getter, ordinal) => getter.getDouble(ordinal)
case d: DecimalType =>
(getter, ordinal) => getter.getDecimal(ordinal, d.precision, d.scale).toString
- case StringType =>
- (getter, ordinal) => new Utf8(getter.getUTF8String(ordinal).getBytes)
- case BinaryType =>
- (getter, ordinal) => ByteBuffer.wrap(getter.getBinary(ordinal))
+ case StringType => avroType.getType match {
+ case Type.ENUM =>
+ import scala.collection.JavaConverters._
+ val enumSymbols: Set[String] = avroType.getEnumSymbols.asScala.toSet
+ (getter, ordinal) =>
+ val data = getter.getUTF8String(ordinal).toString
+ if (!enumSymbols.contains(data)) {
+ throw new IncompatibleSchemaException(
+ "Cannot write \"" + data + "\" since it's not defined in enum \"" +
+ enumSymbols.mkString("\", \""))
--- End diff --
nit: `enumSymbols.mkString("\", \"") + "\"")`
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by lindblombr <gi...@git.apache.org>.
Github user lindblombr commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r205591400
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -165,16 +183,100 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
result
}
- private def resolveNullableType(avroType: Schema, nullable: Boolean): Schema = {
- if (nullable) {
+ private def resolveNullableType(avroType: Schema, catalystType: DataType,
+ nullable: Boolean): Schema = {
+ if (nullable && avroType.getType == Type.UNION) {
// avro uses union to represent nullable type.
- val fields = avroType.getTypes.asScala
- assert(fields.length == 2)
- val actualType = fields.filter(_.getType != NULL)
+ val fieldTypes = avroType.getTypes.asScala
+
+ // If we're nullable, we need to have at least two types. Cases with more than two types are
+ // captured in test("read read-write, read-write w/ schema, read") w/ test.avro input
+ assert(fieldTypes.length >= 2)
+
+ val actualType = catalystType match {
+ case NullType => fieldTypes.filter(_.getType == Type.NULL)
+ case BooleanType => fieldTypes.filter(_.getType == Type.BOOLEAN)
+ case ByteType => fieldTypes.filter(_.getType == Type.INT)
+ case BinaryType =>
+ val at = fieldTypes.filter(x => x.getType == Type.BYTES || x.getType == Type.FIXED)
+ if (at.length > 1) {
+ throw new IncompatibleSchemaException (
+ s"Cannot resolve schema of ${catalystType} against union ${avroType.toString}")
+ } else {
+ at
+ }
+ case ShortType | IntegerType => fieldTypes.filter(_.getType == Type.INT)
+ case LongType => fieldTypes.filter(_.getType == Type.LONG)
+ case FloatType => fieldTypes.filter(_.getType == Type.FLOAT)
+ case DoubleType => fieldTypes.filter(_.getType == Type.DOUBLE)
+ case d: DecimalType => fieldTypes.filter(_.getType == Type.STRING)
+ case StringType => fieldTypes
+ .filter(x => x.getType == Type.STRING || x.getType == Type.ENUM)
+ case DateType => fieldTypes.filter(x => x.getType == Type.INT || x.getType == Type.LONG)
+ case TimestampType => fieldTypes.filter(_.getType == Type.LONG)
+ case ArrayType(et, containsNull) =>
+ // Find array that matches the type
+ fieldTypes.filter(x => x.getType == Type.ARRAY && typeMatchesSchema(et, x.getElementType))
+ case st: StructType => // Find the matching record!
+ val recordTypes = fieldTypes.filter(x => x.getType == Type.RECORD)
+ if (recordTypes.length > 1) {
+ throw new IncompatibleSchemaException(
+ "Unions of multiple record types are NOT supported with user-specified schema")
+ }
+ recordTypes
+ case MapType(kt, vt, valueContainsNull) =>
+ // Find the map that matches the type
+ fieldTypes.filter(x => x.getType == Type.MAP && typeMatchesSchema(vt, x.getValueType))
+ case other =>
+ throw new IncompatibleSchemaException(s"Unexpected type: $other")
+ }
+
assert(actualType.length == 1)
actualType.head
} else {
avroType
}
}
+
+ // Given a Schema and a DataType, do they match?
+ private def typeMatchesSchema(catalystType: DataType, avroSchema: Schema): Boolean = {
+ if (catalystType.isInstanceOf[StructType]) {
+ val avroFields = resolveNullableType(avroSchema, catalystType,
+ avroSchema.getType == Type.UNION)
+ .getFields
+ assert(avroFields.size() == catalystType.asInstanceOf[StructType].length)
+ catalystType.asInstanceOf[StructType].zip(avroFields.asScala).map {
+ case (f1, f2) => typeMatchesSchema(f1.dataType, f2.schema)
+ }.foldLeft(true)(_ && _)
+ } else {
+ val isTypeCompatible = (a: Schema, b: DataType, c: Type) =>
+ resolveNullableType(a, b, a.getType == Type.UNION).getType == c
+
+ catalystType match {
+ case ByteType | ShortType | IntegerType =>
+ isTypeCompatible(avroSchema, catalystType, Type.INT)
+ case BooleanType => isTypeCompatible(avroSchema, catalystType, Type.BOOLEAN)
+ case BinaryType => isTypeCompatible(avroSchema, catalystType, Type.BYTES)
+ case LongType | TimestampType => isTypeCompatible(avroSchema, catalystType, Type.LONG)
+ case FloatType => isTypeCompatible(avroSchema, catalystType, Type.FLOAT)
+ case DoubleType => isTypeCompatible(avroSchema, catalystType, Type.DOUBLE)
+ case d: DecimalType => isTypeCompatible(avroSchema, catalystType, Type.STRING)
--- End diff --
Additional work to support some logical types would probably be in order for this to work correctly. Prior to my change, AvroSerializer always wrote out DecimalType as a string, so I'm just keeping the existing behavior. I added a comment to reflect this.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/94460/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by lindblombr <gi...@git.apache.org>.
Github user lindblombr commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r205912358
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -148,7 +165,8 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
val avroFields = avroStruct.getFields
assert(avroFields.size() == catalystStruct.length)
val fieldConverters = catalystStruct.zip(avroFields.asScala).map {
- case (f1, f2) => newConverter(f1.dataType, resolveNullableType(f2.schema(), f1.nullable))
+ case (f1, f2) => newConverter(f1.dataType, resolveNullableType(
+ f2.schema(), f1.dataType, f1.nullable))
--- End diff --
Done.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/93857/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by lindblombr <gi...@git.apache.org>.
Github user lindblombr commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r206740630
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -165,16 +182,118 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
result
}
- private def resolveNullableType(avroType: Schema, nullable: Boolean): Schema = {
- if (nullable) {
+ // Resolve an Avro union against a supplied DataType, i.e. a LongType compared against
+ // a ["null", "long"] should return a schema of type Schema.Type.LONG
+ // This function also handles resolving a DataType against unions of 2 or more types, i.e.
+ // an IntType resolves against a ["int", "long", "null"] will correctly return a schema of
+ // type Schema.Type.LONG
+ private def resolveUnionType(avroType: Schema, catalystType: DataType,
+ nullable: Boolean): Schema = {
+ if (avroType.getType == Type.UNION) {
// avro uses union to represent nullable type.
- val fields = avroType.getTypes.asScala
- assert(fields.length == 2)
- val actualType = fields.filter(_.getType != NULL)
- assert(actualType.length == 1)
+ val fieldTypes = avroType.getTypes.asScala
+
+ // If we're nullable, we need to have at least two types. Cases with more than two types
+ // are captured in test("read read-write, read-write w/ schema, read") w/ test.avro input
+ if (nullable && fieldTypes.length < 2) {
+ throw new IncompatibleSchemaException(
+ s"Cannot resolve nullable ${catalystType} against union type ${avroType}")
+ }
+
+ val actualType = catalystType match {
+ case NullType => fieldTypes.filter(_.getType == Type.NULL)
+ case BooleanType => fieldTypes.filter(_.getType == Type.BOOLEAN)
+ case ByteType => fieldTypes.filter(_.getType == Type.INT)
+ case BinaryType =>
+ val at = fieldTypes.filter(x => x.getType == Type.BYTES || x.getType == Type.FIXED)
+ if (at.length > 1) {
+ throw new IncompatibleSchemaException(
+ s"Cannot resolve schema of ${catalystType} against union ${avroType.toString}")
+ } else {
+ at
+ }
+ case ShortType | IntegerType => fieldTypes.filter(_.getType == Type.INT)
+ case LongType => fieldTypes.filter(_.getType == Type.LONG)
+ case FloatType => fieldTypes.filter(_.getType == Type.FLOAT)
+ case DoubleType => fieldTypes.filter(_.getType == Type.DOUBLE)
+ case d: DecimalType => fieldTypes.filter(_.getType == Type.STRING)
+ case StringType => fieldTypes
+ .filter(x => x.getType == Type.STRING || x.getType == Type.ENUM)
+ case DateType => fieldTypes.filter(x => x.getType == Type.INT || x.getType == Type.LONG)
+ case TimestampType => fieldTypes.filter(_.getType == Type.LONG)
+ case ArrayType(et, containsNull) =>
+ // Find array that matches the element type specified
+ fieldTypes.filter(x => x.getType == Type.ARRAY
+ && typeMatchesSchema(et, x.getElementType))
+ case st: StructType => // Find the matching record!
+ val recordTypes = fieldTypes.filter(x => x.getType == Type.RECORD)
+ if (recordTypes.length > 1) {
+ throw new IncompatibleSchemaException(
+ "Unions of multiple record types are NOT supported with user-specified schema")
+ }
+ recordTypes
+ case MapType(kt, vt, valueContainsNull) =>
+ // Find the map that matches the value type. Maps in Avro are always key type string
+ fieldTypes.filter(x => x.getType == Type.MAP && typeMatchesSchema(vt, x.getValueType))
--- End diff --
I'm still not clear, but I'll check the behavior prior to this change to see
1. If kt is NOT StringType, was the case handled previously (some string-ification)?
2. If so, I'll update this to match existing behavior
3. If not, it seems we'll have to decide if we want to throw the exception, do the conversion, or something else?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL][WIP]: Built-in AVRO support...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #93565 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93565/testReport)** for PR 21847 at commit [`f05e67e`](https://github.com/apache/spark/commit/f05e67ecc41f38b6625efffd84eea12a356368b9).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by dbtsai <gi...@git.apache.org>.
Github user dbtsai commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r206356380
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -146,9 +159,13 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
private def newStructConverter(
catalystStruct: StructType, avroStruct: Schema): InternalRow => Record = {
val avroFields = avroStruct.getFields
- assert(avroFields.size() == catalystStruct.length)
+ if (avroFields.size != catalystStruct.length) {
+ throw new IncompatibleSchemaException(
+ s"Field list length of ${catalystStruct} does not correspond to length of ${avroStruct}")
--- End diff --
`avroStruct` will print the schema in JSON which will be big and hard to read. `catalystStruct.toString` is not well defined. How about we just print `avroStruct.size` and `catalystStruct.length` with proper message?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by lindblombr <gi...@git.apache.org>.
Github user lindblombr commented on the issue:
https://github.com/apache/spark/pull/21847
RE Performance regression... The test case I used performs the same before the changes in this patch and after, so this doesn't appear related to this change. I'll investigate further and file a Jira if needed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #94523 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/94523/testReport)** for PR 21847 at commit [`0d9c948`](https://github.com/apache/spark/commit/0d9c948eb9b91ed7dcfd2a93ab136b05654eb8f9).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by lindblombr <gi...@git.apache.org>.
Github user lindblombr commented on the issue:
https://github.com/apache/spark/pull/21847
@gengliangwang I think I agree with your take on the multi-type unions in principal. The only issue is that avro itself does support that as a valid use case; however, when I think about the behavior...
If we read in A to SparkSQL where a is type `["null", "int", "long"]`, Spark will automatically up-convert to a long. This will mean that even though the original data may have had a combination of records stored as either "int" or "long", any attempt to write out that same data with user-specified schema will convert all ints to longs, resulting in a slightly different dataset.
This side-effect may be undesirable in some cases (maybe). I personally don't have a use-case for this, except the test data in this module itself includes a multi-type union of this nature, and I wanted this functionality to work for as much of the test data as was available. If, for the sake of simplicity, we'd like to restrict user-specified schemas to two-type unions only, it would still work for my use cases.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by lindblombr <gi...@git.apache.org>.
Github user lindblombr commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r205912718
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -87,17 +88,33 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
case d: DecimalType =>
(getter, ordinal) => getter.getDecimal(ordinal, d.precision, d.scale).toString
case StringType =>
- (getter, ordinal) => new Utf8(getter.getUTF8String(ordinal).getBytes)
+ (getter, ordinal) =>
+ if (avroType.getType == Type.ENUM) {
+ new GenericData.EnumSymbol(avroType, getter.getUTF8String(ordinal).toString)
+ } else {
+ new Utf8(getter.getUTF8String(ordinal).getBytes)
+ }
case BinaryType =>
- (getter, ordinal) => ByteBuffer.wrap(getter.getBinary(ordinal))
+ (getter, ordinal) =>
+ val data = getter.getBinary(ordinal)
+ if (avroType.getType == Type.FIXED) {
+ // Handles fixed-type fields in output schema. Test case is included in test.avro
+ // as it includes several fixed fields that would fail if we specify schema
+ // on-write without this condition
+ val fixed = new GenericData.Fixed(avroType)
+ fixed.bytes(data)
+ fixed
+ } else {
+ ByteBuffer.wrap(data)
+ }
--- End diff --
Thanks for this hint! I reversed the expression as you suggested. I don't think its related to the poor performance of the particular schema I tested, but I think the issue is more general and not related to my changes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #94478 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/94478/testReport)** for PR 21847 at commit [`68ce1b6`](https://github.com/apache/spark/commit/68ce1b69a0a9643c69554a330ee718a4da021c07).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #93858 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93858/testReport)** for PR 21847 at commit [`42978fb`](https://github.com/apache/spark/commit/42978fb13ce5d292f8c4e6a257e6ea14fc8b0670).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by dbtsai <gi...@git.apache.org>.
Github user dbtsai commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r206356838
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -146,9 +159,13 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
private def newStructConverter(
catalystStruct: StructType, avroStruct: Schema): InternalRow => Record = {
val avroFields = avroStruct.getFields
- assert(avroFields.size() == catalystStruct.length)
+ if (avroFields.size != catalystStruct.length) {
+ throw new IncompatibleSchemaException(
+ s"Field list length of ${catalystStruct} does not correspond to length of ${avroStruct}")
+ }
val fieldConverters = catalystStruct.zip(avroFields.asScala).map {
- case (f1, f2) => newConverter(f1.dataType, resolveNullableType(f2.schema(), f1.nullable))
+ case (f1, f2) =>
+ newConverter(f1.dataType, resolveUnionType(f2.schema(), f1.dataType, f1.nullable))
--- End diff --
```scala
val fieldConverters = catalystStruct.zip(avroFields.asScala).map { case (f1, f2) =>
newConverter(f1.dataType, resolveUnionType(f2.schema(), f1.dataType, f1.nullable))
}
```
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #93625 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93625/testReport)** for PR 21847 at commit [`ed44c76`](https://github.com/apache/spark/commit/ed44c76dd2d4abf7d4f6718e90e1a2abe735c9f2).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/93603/
Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL][WIP]: Built-in AVRO support...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #93565 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93565/testReport)** for PR 21847 at commit [`f05e67e`](https://github.com/apache/spark/commit/f05e67ecc41f38b6625efffd84eea12a356368b9).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #93621 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93621/testReport)** for PR 21847 at commit [`621be3e`](https://github.com/apache/spark/commit/621be3edf83742ddb6e14083d9a13217ed7ea19e).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #94523 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/94523/testReport)** for PR 21847 at commit [`0d9c948`](https://github.com/apache/spark/commit/0d9c948eb9b91ed7dcfd2a93ab136b05654eb8f9).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r206368836
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -165,16 +182,118 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
result
}
- private def resolveNullableType(avroType: Schema, nullable: Boolean): Schema = {
- if (nullable) {
+ // Resolve an Avro union against a supplied DataType, i.e. a LongType compared against
+ // a ["null", "long"] should return a schema of type Schema.Type.LONG
+ // This function also handles resolving a DataType against unions of 2 or more types, i.e.
+ // an IntType resolves against a ["int", "long", "null"] will correctly return a schema of
+ // type Schema.Type.LONG
+ private def resolveUnionType(avroType: Schema, catalystType: DataType,
+ nullable: Boolean): Schema = {
+ if (avroType.getType == Type.UNION) {
// avro uses union to represent nullable type.
- val fields = avroType.getTypes.asScala
- assert(fields.length == 2)
- val actualType = fields.filter(_.getType != NULL)
- assert(actualType.length == 1)
+ val fieldTypes = avroType.getTypes.asScala
+
+ // If we're nullable, we need to have at least two types. Cases with more than two types
+ // are captured in test("read read-write, read-write w/ schema, read") w/ test.avro input
+ if (nullable && fieldTypes.length < 2) {
+ throw new IncompatibleSchemaException(
+ s"Cannot resolve nullable ${catalystType} against union type ${avroType}")
+ }
+
+ val actualType = catalystType match {
+ case NullType => fieldTypes.filter(_.getType == Type.NULL)
+ case BooleanType => fieldTypes.filter(_.getType == Type.BOOLEAN)
+ case ByteType => fieldTypes.filter(_.getType == Type.INT)
+ case BinaryType =>
+ val at = fieldTypes.filter(x => x.getType == Type.BYTES || x.getType == Type.FIXED)
+ if (at.length > 1) {
+ throw new IncompatibleSchemaException(
+ s"Cannot resolve schema of ${catalystType} against union ${avroType.toString}")
+ } else {
+ at
+ }
+ case ShortType | IntegerType => fieldTypes.filter(_.getType == Type.INT)
+ case LongType => fieldTypes.filter(_.getType == Type.LONG)
+ case FloatType => fieldTypes.filter(_.getType == Type.FLOAT)
+ case DoubleType => fieldTypes.filter(_.getType == Type.DOUBLE)
+ case d: DecimalType => fieldTypes.filter(_.getType == Type.STRING)
+ case StringType => fieldTypes
+ .filter(x => x.getType == Type.STRING || x.getType == Type.ENUM)
+ case DateType => fieldTypes.filter(x => x.getType == Type.INT || x.getType == Type.LONG)
+ case TimestampType => fieldTypes.filter(_.getType == Type.LONG)
+ case ArrayType(et, containsNull) =>
+ // Find array that matches the element type specified
+ fieldTypes.filter(x => x.getType == Type.ARRAY
+ && typeMatchesSchema(et, x.getElementType))
+ case st: StructType => // Find the matching record!
+ val recordTypes = fieldTypes.filter(x => x.getType == Type.RECORD)
+ if (recordTypes.length > 1) {
+ throw new IncompatibleSchemaException(
+ "Unions of multiple record types are NOT supported with user-specified schema")
+ }
+ recordTypes
+ case MapType(kt, vt, valueContainsNull) =>
+ // Find the map that matches the value type. Maps in Avro are always key type string
+ fieldTypes.filter(x => x.getType == Type.MAP && typeMatchesSchema(vt, x.getValueType))
--- End diff --
Does this mean no matter `kt` is any type, the key type of map in Avro is always string?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by dbtsai <gi...@git.apache.org>.
Github user dbtsai commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r205683257
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -148,7 +165,8 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
val avroFields = avroStruct.getFields
assert(avroFields.size() == catalystStruct.length)
val fieldConverters = catalystStruct.zip(avroFields.asScala).map {
- case (f1, f2) => newConverter(f1.dataType, resolveNullableType(f2.schema(), f1.nullable))
+ case (f1, f2) => newConverter(f1.dataType, resolveNullableType(
+ f2.schema(), f1.dataType, f1.nullable))
--- End diff --
Nit, formating,
```scala
case (f1, f2) =>
newConverter(f1.dataType, resolveNullableType(f2.schema(), f1.dataType, f1.nullable))
```
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r209115909
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -87,10 +87,36 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
(getter, ordinal) => getter.getDouble(ordinal)
case d: DecimalType =>
(getter, ordinal) => getter.getDecimal(ordinal, d.precision, d.scale).toString
- case StringType =>
- (getter, ordinal) => new Utf8(getter.getUTF8String(ordinal).getBytes)
- case BinaryType =>
- (getter, ordinal) => ByteBuffer.wrap(getter.getBinary(ordinal))
+ case StringType => avroType.getType match {
+ case Type.ENUM =>
+ import scala.collection.JavaConverters._
+ val enumSymbols: Set[String] = avroType.getEnumSymbols.asScala.toSet
+ (getter, ordinal) =>
+ val data = getter.getUTF8String(ordinal).toString
+ if (!enumSymbols.contains(data)) {
+ throw new IncompatibleSchemaException(
+ "Cannot write \"" + data + "\" since it's not defined in enum \"" +
+ enumSymbols.mkString("\", \""))
+ }
+ new EnumSymbol(avroType, data)
+ case _ =>
+ (getter, ordinal) => new Utf8(getter.getUTF8String(ordinal).getBytes)
+ }
+ case BinaryType => avroType.getType match {
+ case Type.FIXED =>
+ val size = avroType.getFixedSize()
+ (getter, ordinal) =>
+ val data: Array[Byte] = getter.getBinary(ordinal)
+ if (data.length != size) {
+ throw new IncompatibleSchemaException(
+ s"Cannot write ${data.length} ${if (data.length > 1) "bytes" else "byte"} of " +
+ s"binary data into FIXED Type with size of " +
--- End diff --
nit: `"binary data into FIXED Type with size of "`.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #93626 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93626/testReport)** for PR 21847 at commit [`c76aadd`](https://github.com/apache/spark/commit/c76aaddb40275537dded99ab0c708a6802e10a30).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by lindblombr <gi...@git.apache.org>.
Github user lindblombr commented on the issue:
https://github.com/apache/spark/pull/21847
@MaxGekk thanks for the feedback! Should be addressed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by dbtsai <gi...@git.apache.org>.
Github user dbtsai commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r209117064
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -87,10 +87,36 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
(getter, ordinal) => getter.getDouble(ordinal)
case d: DecimalType =>
(getter, ordinal) => getter.getDecimal(ordinal, d.precision, d.scale).toString
- case StringType =>
- (getter, ordinal) => new Utf8(getter.getUTF8String(ordinal).getBytes)
- case BinaryType =>
- (getter, ordinal) => ByteBuffer.wrap(getter.getBinary(ordinal))
+ case StringType => avroType.getType match {
+ case Type.ENUM =>
+ import scala.collection.JavaConverters._
+ val enumSymbols: Set[String] = avroType.getEnumSymbols.asScala.toSet
+ (getter, ordinal) =>
+ val data = getter.getUTF8String(ordinal).toString
+ if (!enumSymbols.contains(data)) {
+ throw new IncompatibleSchemaException(
+ "Cannot write \"" + data + "\" since it's not defined in enum \"" +
+ enumSymbols.mkString("\", \""))
+ }
+ new EnumSymbol(avroType, data)
+ case _ =>
+ (getter, ordinal) => new Utf8(getter.getUTF8String(ordinal).getBytes)
+ }
+ case BinaryType => avroType.getType match {
+ case Type.FIXED =>
+ val size = avroType.getFixedSize()
+ (getter, ordinal) =>
+ val data: Array[Byte] = getter.getBinary(ordinal)
+ if (data.length != size) {
+ throw new IncompatibleSchemaException(
+ s"Cannot write ${data.length} ${if (data.length > 1) "bytes" else "byte"} of " +
+ s"binary data into FIXED Type with size of " +
--- End diff --
addressed
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #94166 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/94166/testReport)** for PR 21847 at commit [`eebc16d`](https://github.com/apache/spark/commit/eebc16dec56aca6ae7086906ba43f0e9896075a3).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL][WIP]: Built-in AVRO ...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r205236872
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -41,7 +42,7 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
}
private val converter: Any => Any = {
- val actualAvroType = resolveNullableType(rootAvroType, nullable)
+ val actualAvroType = resolveNullableType(rootAvroType, rootCatalystType, nullable)
--- End diff --
Is the change here related to specifying schema on write?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL][WIP]: Built-in AVRO support...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #93457 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93457/testReport)** for PR 21847 at commit [`71dbc39`](https://github.com/apache/spark/commit/71dbc398846b073143bea3e5c038ea171cf7aa50).
* This patch **fails to build**.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by dbtsai <gi...@git.apache.org>.
Github user dbtsai commented on the issue:
https://github.com/apache/spark/pull/21847
+cc @MaxGekk and @gengliangwang who worked on this part of codebase.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/94522/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL][WIP]: Built-in AVRO support...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #93464 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93464/testReport)** for PR 21847 at commit [`12b3859`](https://github.com/apache/spark/commit/12b3859467155e2e205afffcc7f4db9def282bab).
* This patch **fails to build**.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL][WIP]: Built-in AVRO support...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Merged build finished. Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/94524/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r205550953
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -165,16 +183,100 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
result
}
- private def resolveNullableType(avroType: Schema, nullable: Boolean): Schema = {
- if (nullable) {
+ private def resolveNullableType(avroType: Schema, catalystType: DataType,
+ nullable: Boolean): Schema = {
+ if (nullable && avroType.getType == Type.UNION) {
--- End diff --
Is it possible `nullable == false` but `avroType.getType == Type.UNION`?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #94460 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/94460/testReport)** for PR 21847 at commit [`46fec45`](https://github.com/apache/spark/commit/46fec4519267ef28824c7745fa30f1c1f6a7b056).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #93625 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93625/testReport)** for PR 21847 at commit [`ed44c76`](https://github.com/apache/spark/commit/ed44c76dd2d4abf7d4f6718e90e1a2abe735c9f2).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/94166/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/spark/pull/21847
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL][WIP]: Built-in AVRO support...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #93457 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93457/testReport)** for PR 21847 at commit [`71dbc39`](https://github.com/apache/spark/commit/71dbc398846b073143bea3e5c038ea171cf7aa50).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL][WIP]: Built-in AVRO ...
Posted by lindblombr <gi...@git.apache.org>.
Github user lindblombr commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r205505006
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -41,7 +42,7 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
}
private val converter: Any => Any = {
- val actualAvroType = resolveNullableType(rootAvroType, nullable)
+ val actualAvroType = resolveNullableType(rootAvroType, rootCatalystType, nullable)
--- End diff --
Yes. I need the original `rootCatalystType` so I can determine how those types map back to the specified schema. Simply having `nullable` as context is not sufficient :(
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/93846/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/94190/
Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #93707 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93707/testReport)** for PR 21847 at commit [`874d2a8`](https://github.com/apache/spark/commit/874d2a86369083ba271e737b36f19baf2710ee01).
* This patch **fails Scala style tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/93619/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #93619 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93619/testReport)** for PR 21847 at commit [`d85242e`](https://github.com/apache/spark/commit/d85242e1dd1b24847dc1fca8c8040cb72e9fab53).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by MaxGekk <gi...@git.apache.org>.
Github user MaxGekk commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r205562110
--- Diff: external/avro/src/test/resources/multirecordtypeunion.avsc ---
@@ -0,0 +1,31 @@
+{
+ "name": "MultipleRecordTypesUnion",
+ "namespace": "com.apple.siri.loggingSchemas.schemas",
--- End diff --
I hope you didn't expose your internal info.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r208841399
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -87,10 +87,18 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
(getter, ordinal) => getter.getDouble(ordinal)
case d: DecimalType =>
(getter, ordinal) => getter.getDecimal(ordinal, d.precision, d.scale).toString
- case StringType =>
- (getter, ordinal) => new Utf8(getter.getUTF8String(ordinal).getBytes)
- case BinaryType =>
- (getter, ordinal) => ByteBuffer.wrap(getter.getBinary(ordinal))
+ case StringType => avroType.getType match {
+ case Type.ENUM =>
+ (getter, ordinal) => new EnumSymbol(avroType, getter.getUTF8String(ordinal).toString)
+ case _ =>
+ (getter, ordinal) => new Utf8(getter.getUTF8String(ordinal).getBytes)
+ }
+ case BinaryType => avroType.getType match {
+ case Type.FIXED =>
--- End diff --
FIXED has a "size" attribute, shall we consider it when preparing the bytes? e.g. shall we throw exception if the bytes from Spark exceed the size, and shall we padding the bytes when its length is smaller than the size.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by lindblombr <gi...@git.apache.org>.
Github user lindblombr commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r206691906
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -87,17 +87,30 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
case d: DecimalType =>
(getter, ordinal) => getter.getDecimal(ordinal, d.precision, d.scale).toString
case StringType =>
- (getter, ordinal) => new Utf8(getter.getUTF8String(ordinal).getBytes)
+ if (avroType.getType == Type.ENUM) {
+ (getter, ordinal) =>
+ new EnumSymbol(avroType, getter.getUTF8String(ordinal).toString)
+ } else {
+ (getter, ordinal) =>
+ new Utf8(getter.getUTF8String(ordinal).getBytes)
+ }
case BinaryType =>
- (getter, ordinal) => ByteBuffer.wrap(getter.getBinary(ordinal))
+ if (avroType.getType == Type.FIXED) {
+ // Handles fixed-type fields in output schema. Test case is included in test.avro
+ // as it includes several fixed fields that would fail if we specify schema
+ // on-write without this condition
+ (getter, ordinal) => new Fixed(avroType, getter.getBinary(ordinal))
+ } else {
+ (getter, ordinal) => ByteBuffer.wrap(getter.getBinary(ordinal))
+ }
case DateType =>
(getter, ordinal) => getter.getInt(ordinal) * DateTimeUtils.MILLIS_PER_DAY
case TimestampType =>
(getter, ordinal) => getter.getLong(ordinal) / 1000
case ArrayType(et, containsNull) =>
val elementConverter = newConverter(
- et, resolveNullableType(avroType.getElementType, containsNull))
+ et, resolveUnionType(avroType.getElementType, et, containsNull))
(getter, ordinal) => {
--- End diff --
Done.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL][WIP]: Built-in AVRO support...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/93453/
Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #93699 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93699/testReport)** for PR 21847 at commit [`1671ade`](https://github.com/apache/spark/commit/1671adeac3d306372f213794e93396c2bc57776f).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/93707/
Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by dbtsai <gi...@git.apache.org>.
Github user dbtsai commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r207358817
--- Diff: external/avro/src/test/resources/messy.avsc ---
@@ -0,0 +1,140 @@
+{
--- End diff --
Instead of adding this complicated avro json schema, should we test smaller cases as unit tests in `AvroCatalystDataConversionSuite` so in the future, it will be easier to know what is broken? Thanks.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL][WIP]: Built-in AVRO support...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Merged build finished. Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by MaxGekk <gi...@git.apache.org>.
Github user MaxGekk commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r205560983
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -165,16 +183,100 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
result
}
- private def resolveNullableType(avroType: Schema, nullable: Boolean): Schema = {
- if (nullable) {
+ private def resolveNullableType(avroType: Schema, catalystType: DataType,
+ nullable: Boolean): Schema = {
+ if (nullable && avroType.getType == Type.UNION) {
// avro uses union to represent nullable type.
- val fields = avroType.getTypes.asScala
- assert(fields.length == 2)
- val actualType = fields.filter(_.getType != NULL)
+ val fieldTypes = avroType.getTypes.asScala
+
+ // If we're nullable, we need to have at least two types. Cases with more than two types are
+ // captured in test("read read-write, read-write w/ schema, read") w/ test.avro input
+ assert(fieldTypes.length >= 2)
+
+ val actualType = catalystType match {
+ case NullType => fieldTypes.filter(_.getType == Type.NULL)
+ case BooleanType => fieldTypes.filter(_.getType == Type.BOOLEAN)
+ case ByteType => fieldTypes.filter(_.getType == Type.INT)
+ case BinaryType =>
+ val at = fieldTypes.filter(x => x.getType == Type.BYTES || x.getType == Type.FIXED)
+ if (at.length > 1) {
+ throw new IncompatibleSchemaException (
+ s"Cannot resolve schema of ${catalystType} against union ${avroType.toString}")
+ } else {
+ at
+ }
+ case ShortType | IntegerType => fieldTypes.filter(_.getType == Type.INT)
+ case LongType => fieldTypes.filter(_.getType == Type.LONG)
+ case FloatType => fieldTypes.filter(_.getType == Type.FLOAT)
+ case DoubleType => fieldTypes.filter(_.getType == Type.DOUBLE)
+ case d: DecimalType => fieldTypes.filter(_.getType == Type.STRING)
+ case StringType => fieldTypes
+ .filter(x => x.getType == Type.STRING || x.getType == Type.ENUM)
+ case DateType => fieldTypes.filter(x => x.getType == Type.INT || x.getType == Type.LONG)
+ case TimestampType => fieldTypes.filter(_.getType == Type.LONG)
+ case ArrayType(et, containsNull) =>
+ // Find array that matches the type
+ fieldTypes.filter(x => x.getType == Type.ARRAY && typeMatchesSchema(et, x.getElementType))
+ case st: StructType => // Find the matching record!
+ val recordTypes = fieldTypes.filter(x => x.getType == Type.RECORD)
+ if (recordTypes.length > 1) {
+ throw new IncompatibleSchemaException(
+ "Unions of multiple record types are NOT supported with user-specified schema")
+ }
+ recordTypes
+ case MapType(kt, vt, valueContainsNull) =>
+ // Find the map that matches the type
+ fieldTypes.filter(x => x.getType == Type.MAP && typeMatchesSchema(vt, x.getValueType))
+ case other =>
+ throw new IncompatibleSchemaException(s"Unexpected type: $other")
+ }
+
assert(actualType.length == 1)
actualType.head
} else {
avroType
}
}
+
+ // Given a Schema and a DataType, do they match?
+ private def typeMatchesSchema(catalystType: DataType, avroSchema: Schema): Boolean = {
+ if (catalystType.isInstanceOf[StructType]) {
+ val avroFields = resolveNullableType(avroSchema, catalystType,
+ avroSchema.getType == Type.UNION)
+ .getFields
+ assert(avroFields.size() == catalystType.asInstanceOf[StructType].length)
+ catalystType.asInstanceOf[StructType].zip(avroFields.asScala).map {
+ case (f1, f2) => typeMatchesSchema(f1.dataType, f2.schema)
+ }.foldLeft(true)(_ && _)
--- End diff --
`map{...}.foldLeft(true)(_ && _)` -> `forall{...}`
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by dbtsai <gi...@git.apache.org>.
Github user dbtsai commented on the issue:
https://github.com/apache/spark/pull/21847
Thanks all. Merged into master.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by MaxGekk <gi...@git.apache.org>.
Github user MaxGekk commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r205560005
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -165,16 +183,100 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
result
}
- private def resolveNullableType(avroType: Schema, nullable: Boolean): Schema = {
- if (nullable) {
+ private def resolveNullableType(avroType: Schema, catalystType: DataType,
+ nullable: Boolean): Schema = {
+ if (nullable && avroType.getType == Type.UNION) {
// avro uses union to represent nullable type.
- val fields = avroType.getTypes.asScala
- assert(fields.length == 2)
- val actualType = fields.filter(_.getType != NULL)
+ val fieldTypes = avroType.getTypes.asScala
+
+ // If we're nullable, we need to have at least two types. Cases with more than two types are
+ // captured in test("read read-write, read-write w/ schema, read") w/ test.avro input
+ assert(fieldTypes.length >= 2)
+
+ val actualType = catalystType match {
+ case NullType => fieldTypes.filter(_.getType == Type.NULL)
+ case BooleanType => fieldTypes.filter(_.getType == Type.BOOLEAN)
+ case ByteType => fieldTypes.filter(_.getType == Type.INT)
+ case BinaryType =>
+ val at = fieldTypes.filter(x => x.getType == Type.BYTES || x.getType == Type.FIXED)
+ if (at.length > 1) {
+ throw new IncompatibleSchemaException (
+ s"Cannot resolve schema of ${catalystType} against union ${avroType.toString}")
+ } else {
+ at
+ }
+ case ShortType | IntegerType => fieldTypes.filter(_.getType == Type.INT)
+ case LongType => fieldTypes.filter(_.getType == Type.LONG)
+ case FloatType => fieldTypes.filter(_.getType == Type.FLOAT)
+ case DoubleType => fieldTypes.filter(_.getType == Type.DOUBLE)
+ case d: DecimalType => fieldTypes.filter(_.getType == Type.STRING)
+ case StringType => fieldTypes
+ .filter(x => x.getType == Type.STRING || x.getType == Type.ENUM)
+ case DateType => fieldTypes.filter(x => x.getType == Type.INT || x.getType == Type.LONG)
+ case TimestampType => fieldTypes.filter(_.getType == Type.LONG)
+ case ArrayType(et, containsNull) =>
+ // Find array that matches the type
+ fieldTypes.filter(x => x.getType == Type.ARRAY && typeMatchesSchema(et, x.getElementType))
+ case st: StructType => // Find the matching record!
+ val recordTypes = fieldTypes.filter(x => x.getType == Type.RECORD)
+ if (recordTypes.length > 1) {
+ throw new IncompatibleSchemaException(
+ "Unions of multiple record types are NOT supported with user-specified schema")
+ }
+ recordTypes
+ case MapType(kt, vt, valueContainsNull) =>
+ // Find the map that matches the type
+ fieldTypes.filter(x => x.getType == Type.MAP && typeMatchesSchema(vt, x.getValueType))
+ case other =>
+ throw new IncompatibleSchemaException(s"Unexpected type: $other")
+ }
+
assert(actualType.length == 1)
actualType.head
} else {
avroType
}
}
+
+ // Given a Schema and a DataType, do they match?
+ private def typeMatchesSchema(catalystType: DataType, avroSchema: Schema): Boolean = {
+ if (catalystType.isInstanceOf[StructType]) {
+ val avroFields = resolveNullableType(avroSchema, catalystType,
+ avroSchema.getType == Type.UNION)
+ .getFields
+ assert(avroFields.size() == catalystType.asInstanceOf[StructType].length)
--- End diff --
Why don't you return `false` instead of crashing in the assert in the case of different sizes?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on the issue:
https://github.com/apache/spark/pull/21847
LGTM except one concern about error checking for fixed type. Thanks for working on it!
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/94531/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by dbtsai <gi...@git.apache.org>.
Github user dbtsai commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r205692778
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -165,16 +183,112 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
result
}
- private def resolveNullableType(avroType: Schema, nullable: Boolean): Schema = {
- if (nullable) {
- // avro uses union to represent nullable type.
- val fields = avroType.getTypes.asScala
- assert(fields.length == 2)
- val actualType = fields.filter(_.getType != NULL)
- assert(actualType.length == 1)
- actualType.head
+ // Resolve an Avro union against a supplied DataType, i.e. a LongType compared against
+ // a ["null", "long"] should return a schema of type Schema.Type.LONG
+ // This function also handles resolving a DataType against unions of 2 or more types, i.e.
+ // an IntType resolves against a ["int", "long", "null"] will correctly return a schema of
+ // type Schema.Type.LONG
+ private def resolveNullableType(avroType: Schema, catalystType: DataType,
+ nullable: Boolean): Schema = {
+ (nullable, avroType.getType) match {
+ case (false, Type.UNION) | (true, Type.UNION) =>
+ // avro uses union to represent nullable type.
+ val fieldTypes = avroType.getTypes.asScala
+
+ // If we're nullable, we need to have at least two types. Cases with more than two types
+ // are captured in test("read read-write, read-write w/ schema, read") w/ test.avro input
+ assert(fieldTypes.length >= 2)
+
+ val actualType = catalystType match {
+ case NullType => fieldTypes.filter(_.getType == Type.NULL)
+ case BooleanType => fieldTypes.filter(_.getType == Type.BOOLEAN)
+ case ByteType => fieldTypes.filter(_.getType == Type.INT)
+ case BinaryType =>
+ val at = fieldTypes.filter(x => x.getType == Type.BYTES || x.getType == Type.FIXED)
+ if (at.length > 1) {
+ throw new IncompatibleSchemaException(
+ s"Cannot resolve schema of ${catalystType} against union ${avroType.toString}")
+ } else {
+ at
+ }
+ case ShortType | IntegerType => fieldTypes.filter(_.getType == Type.INT)
+ case LongType => fieldTypes.filter(_.getType == Type.LONG)
+ case FloatType => fieldTypes.filter(_.getType == Type.FLOAT)
+ case DoubleType => fieldTypes.filter(_.getType == Type.DOUBLE)
+ case d: DecimalType => fieldTypes.filter(_.getType == Type.STRING)
+ case StringType => fieldTypes
+ .filter(x => x.getType == Type.STRING || x.getType == Type.ENUM)
+ case DateType => fieldTypes.filter(x => x.getType == Type.INT || x.getType == Type.LONG)
+ case TimestampType => fieldTypes.filter(_.getType == Type.LONG)
+ case ArrayType(et, containsNull) =>
+ // Find array that matches the element type specified
+ fieldTypes.filter(x => x.getType == Type.ARRAY
+ && typeMatchesSchema(et, x.getElementType))
+ case st: StructType => // Find the matching record!
+ val recordTypes = fieldTypes.filter(x => x.getType == Type.RECORD)
+ if (recordTypes.length > 1) {
+ throw new IncompatibleSchemaException(
+ "Unions of multiple record types are NOT supported with user-specified schema")
+ }
+ recordTypes
+ case MapType(kt, vt, valueContainsNull) =>
+ // Find the map that matches the value type. Maps in Avro are always key type string
+ fieldTypes.filter(x => x.getType == Type.MAP && typeMatchesSchema(vt, x.getValueType))
+ case other =>
+ throw new IncompatibleSchemaException(s"Unexpected type: $other")
+ }
+
+ assert(actualType.length == 1)
--- End diff --
Can you elaborate when `actualType.length == 0` or `actualType.length > 1`?
Is it possible that `catalystType` is `Int`, and `fieldTypes` only contains `Long`? Do we want to do the promotion?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by lindblombr <gi...@git.apache.org>.
Github user lindblombr commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r205568259
--- Diff: external/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala ---
@@ -40,12 +41,40 @@ import org.apache.spark.sql.types._
class AvroSuite extends QueryTest with SharedSQLContext with SQLTestUtils {
val episodesAvro = testFile("episodes.avro")
val testAvro = testFile("test.avro")
+ val messyAvro = testFile("messy.avro")
+ val multiRecordTypeUnionAvro = testFile("multirecordtypeunion.avro")
+ val episodesSchemaFile = testFile("episodes.avsc")
+ val testSchemaFile = testFile("test.avsc")
+ val messySchemaFile = testFile("messy.avsc")
+ val multiRecordTypeUnionSchemaFile = testFile("multirecordtypeunion.avsc")
override protected def beforeAll(): Unit = {
super.beforeAll()
spark.conf.set("spark.sql.files.maxPartitionBytes", 1024)
}
+ def forceSchemaCheck(spark: SparkSession, inputPath: String, schemaFile: String): Unit = {
+ withTempPath { tempDir =>
+ val df = spark.read.format("avro").load(inputPath)
+
+ val tempSaveDir1 = s"$tempDir/test1/"
+ val tempSaveDir2 = s"$tempDir/test2/"
+
+ df.write.format("avro").save(tempSaveDir1)
+
+ val newDf = spark.read.format("avro").load(tempSaveDir1)
+ checkAnswer(df, newDf)
--- End diff --
I changed from `forceSchemaCheck` to `checkSpecifySchemaOnWrite`. I think thats a bit more clear.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r205549727
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -165,16 +183,100 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
result
}
- private def resolveNullableType(avroType: Schema, nullable: Boolean): Schema = {
- if (nullable) {
+ private def resolveNullableType(avroType: Schema, catalystType: DataType,
+ nullable: Boolean): Schema = {
+ if (nullable && avroType.getType == Type.UNION) {
// avro uses union to represent nullable type.
- val fields = avroType.getTypes.asScala
- assert(fields.length == 2)
- val actualType = fields.filter(_.getType != NULL)
+ val fieldTypes = avroType.getTypes.asScala
+
+ // If we're nullable, we need to have at least two types. Cases with more than two types are
+ // captured in test("read read-write, read-write w/ schema, read") w/ test.avro input
+ assert(fieldTypes.length >= 2)
+
+ val actualType = catalystType match {
+ case NullType => fieldTypes.filter(_.getType == Type.NULL)
+ case BooleanType => fieldTypes.filter(_.getType == Type.BOOLEAN)
+ case ByteType => fieldTypes.filter(_.getType == Type.INT)
+ case BinaryType =>
+ val at = fieldTypes.filter(x => x.getType == Type.BYTES || x.getType == Type.FIXED)
+ if (at.length > 1) {
+ throw new IncompatibleSchemaException (
+ s"Cannot resolve schema of ${catalystType} against union ${avroType.toString}")
+ } else {
+ at
+ }
+ case ShortType | IntegerType => fieldTypes.filter(_.getType == Type.INT)
+ case LongType => fieldTypes.filter(_.getType == Type.LONG)
+ case FloatType => fieldTypes.filter(_.getType == Type.FLOAT)
+ case DoubleType => fieldTypes.filter(_.getType == Type.DOUBLE)
+ case d: DecimalType => fieldTypes.filter(_.getType == Type.STRING)
+ case StringType => fieldTypes
+ .filter(x => x.getType == Type.STRING || x.getType == Type.ENUM)
+ case DateType => fieldTypes.filter(x => x.getType == Type.INT || x.getType == Type.LONG)
+ case TimestampType => fieldTypes.filter(_.getType == Type.LONG)
+ case ArrayType(et, containsNull) =>
+ // Find array that matches the type
+ fieldTypes.filter(x => x.getType == Type.ARRAY && typeMatchesSchema(et, x.getElementType))
+ case st: StructType => // Find the matching record!
+ val recordTypes = fieldTypes.filter(x => x.getType == Type.RECORD)
+ if (recordTypes.length > 1) {
+ throw new IncompatibleSchemaException(
+ "Unions of multiple record types are NOT supported with user-specified schema")
+ }
+ recordTypes
+ case MapType(kt, vt, valueContainsNull) =>
+ // Find the map that matches the type
+ fieldTypes.filter(x => x.getType == Type.MAP && typeMatchesSchema(vt, x.getValueType))
+ case other =>
+ throw new IncompatibleSchemaException(s"Unexpected type: $other")
+ }
+
assert(actualType.length == 1)
actualType.head
} else {
avroType
}
}
+
+ // Given a Schema and a DataType, do they match?
+ private def typeMatchesSchema(catalystType: DataType, avroSchema: Schema): Boolean = {
+ if (catalystType.isInstanceOf[StructType]) {
+ val avroFields = resolveNullableType(avroSchema, catalystType,
+ avroSchema.getType == Type.UNION)
+ .getFields
+ assert(avroFields.size() == catalystType.asInstanceOf[StructType].length)
+ catalystType.asInstanceOf[StructType].zip(avroFields.asScala).map {
+ case (f1, f2) => typeMatchesSchema(f1.dataType, f2.schema)
+ }.foldLeft(true)(_ && _)
--- End diff --
We can save `foldLeft` here?
```scala
catalystType.asInstanceOf[StructType].zip(avroFields.asScala).forall {
case (f1, f2) => typeMatchesSchema(f1.dataType, f2.schema)
}
```
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by lindblombr <gi...@git.apache.org>.
Github user lindblombr commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r205592450
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -165,16 +183,100 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
result
}
- private def resolveNullableType(avroType: Schema, nullable: Boolean): Schema = {
- if (nullable) {
+ private def resolveNullableType(avroType: Schema, catalystType: DataType,
+ nullable: Boolean): Schema = {
+ if (nullable && avroType.getType == Type.UNION) {
--- End diff --
I've converted this into a match statement that covers the four cases:
* nullable == false and Type.UNION => should "resolve" the union to the appropriate type
* nullable == true and Type.UNION => should "resolve" the union to the appropriate type
* nullable == Any and Any (Type) => just return the Type
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r208842200
--- Diff: external/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala ---
@@ -725,6 +744,158 @@ class AvroSuite extends QueryTest with SharedSQLContext with SQLTestUtils {
assert(result === Row("foo"))
}
+ test("support user provided avro schema for writing nullable enum type") {
+ withTempPath { tempDir =>
+ val avroSchema =
+ """
+ |{
+ | "type" : "record",
+ | "name" : "test_schema",
+ | "fields" : [{
+ | "name": "enum",
+ | "type": [{ "type": "enum",
+ | "name": "Suit",
+ | "symbols" : ["SPADES", "HEARTS", "DIAMONDS", "CLUBS"]
+ | }, "null"]
+ | }]
+ |}
+ """.stripMargin
+
+ val df = spark.createDataFrame(spark.sparkContext.parallelize(Seq(
+ Row("SPADES"), Row(null), Row("HEARTS"), Row("DIAMONDS"),
+ Row(null), Row("CLUBS"), Row("HEARTS"), Row("SPADES"))),
+ StructType(Seq(StructField("Suit", StringType, true))))
+
+ val tempSaveDir = s"$tempDir/save/"
+
+ df.write.format("avro").option("avroSchema", avroSchema).save(tempSaveDir)
+
+ checkAnswer(df, spark.read.format("avro").load(tempSaveDir))
+ checkAvroSchemaEquals(avroSchema, getAvroSchemaStringFromFiles(tempSaveDir))
+
+ // Writing df containing data not in the enum will throw an exception
+ intercept[SparkException] {
--- End diff --
can we also check the error message?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on the issue:
https://github.com/apache/spark/pull/21847
LGTM
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by gengliangwang <gi...@git.apache.org>.
Github user gengliangwang commented on the issue:
https://github.com/apache/spark/pull/21847
Since the data type in Spark is certain, why do we need to support output Avro schema like `["int", "long", "null"]`?
Can we just forbid such usage by having a rule for the schema: If the Avro type is Union, it has at most two types and one of it is Null type.
Otherwise things can be complicated.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #94478 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/94478/testReport)** for PR 21847 at commit [`68ce1b6`](https://github.com/apache/spark/commit/68ce1b69a0a9643c69554a330ee718a4da021c07).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by lindblombr <gi...@git.apache.org>.
Github user lindblombr commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r208770874
--- Diff: external/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala ---
@@ -78,6 +79,25 @@ class AvroSuite extends QueryTest with SharedSQLContext with SQLTestUtils {
checkAnswer(newEntries, originalEntries)
}
+ def checkAvroSchemaEquals(avroSchema: String, expectedAvroSchema: String): Unit = {
+ assert(new Schema.Parser().parse(avroSchema) ==
+ new Schema.Parser().parse(expectedAvroSchema))
+ }
+
+ def getAvroSchemaStringFromFiles(filePath: String): String = {
--- End diff --
Should we update the test L980 `"Validate namespace in avro file that has nested records with the same name"` to use this, too?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL][WIP]: Built-in AVRO support...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #93453 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93453/testReport)** for PR 21847 at commit [`30fc1ae`](https://github.com/apache/spark/commit/30fc1ae98a2c29824b83e956c2cbe87e9bbb3749).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r206699403
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -165,16 +186,117 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
result
}
- private def resolveNullableType(avroType: Schema, nullable: Boolean): Schema = {
- if (nullable) {
+ // Resolve an Avro union against a supplied DataType, i.e. a LongType compared against
+ // a ["null", "long"] should return a schema of type Schema.Type.LONG
+ // This function also handles resolving a DataType against unions of 2 or more types, i.e.
+ // an IntType resolves against a ["int", "long", "null"] will correctly return a schema of
+ // type Schema.Type.LONG
+ private def resolveUnionType(avroType: Schema, catalystType: DataType,
+ nullable: Boolean): Schema = {
+ if (avroType.getType == Type.UNION) {
// avro uses union to represent nullable type.
- val fields = avroType.getTypes.asScala
- assert(fields.length == 2)
- val actualType = fields.filter(_.getType != NULL)
- assert(actualType.length == 1)
+ val fieldTypes = avroType.getTypes.asScala
+
+ // If we're nullable, we need to have at least two types. Cases with more than two types
+ // are captured in test("read read-write, read-write w/ schema, read") w/ test.avro input
+ if (nullable && fieldTypes.length < 2) {
+ throw new IncompatibleSchemaException(
+ s"Cannot resolve nullable ${catalystType} against union type ${avroType}")
+ }
+
+ val actualType = catalystType match {
+ case NullType => fieldTypes.filter(_.getType == Type.NULL)
+ case BooleanType => fieldTypes.filter(_.getType == Type.BOOLEAN)
+ case ByteType => fieldTypes.filter(_.getType == Type.INT)
+ case BinaryType => fieldTypes
+ .filter(x => x.getType == Type.BYTES || x.getType == Type.FIXED)
+ case ShortType | IntegerType => fieldTypes.filter(_.getType == Type.INT)
+ case LongType => fieldTypes.filter(_.getType == Type.LONG)
+ case FloatType => fieldTypes.filter(_.getType == Type.FLOAT)
+ case DoubleType => fieldTypes.filter(_.getType == Type.DOUBLE)
+ case d: DecimalType => fieldTypes.filter(_.getType == Type.STRING)
+ case StringType => fieldTypes
+ .filter(x => x.getType == Type.STRING || x.getType == Type.ENUM)
+ case DateType => fieldTypes
+ .filter(x => x.getType == Type.INT || x.getType == Type.LONG)
--- End diff --
I've asked in previous comment: Why we need consider long? I think `DateType` uses int?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #93621 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93621/testReport)** for PR 21847 at commit [`621be3e`](https://github.com/apache/spark/commit/621be3edf83742ddb6e14083d9a13217ed7ea19e).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by lindblombr <gi...@git.apache.org>.
Github user lindblombr commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r205912517
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -165,16 +183,112 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
result
}
- private def resolveNullableType(avroType: Schema, nullable: Boolean): Schema = {
- if (nullable) {
- // avro uses union to represent nullable type.
- val fields = avroType.getTypes.asScala
- assert(fields.length == 2)
- val actualType = fields.filter(_.getType != NULL)
- assert(actualType.length == 1)
- actualType.head
+ // Resolve an Avro union against a supplied DataType, i.e. a LongType compared against
+ // a ["null", "long"] should return a schema of type Schema.Type.LONG
+ // This function also handles resolving a DataType against unions of 2 or more types, i.e.
+ // an IntType resolves against a ["int", "long", "null"] will correctly return a schema of
+ // type Schema.Type.LONG
+ private def resolveNullableType(avroType: Schema, catalystType: DataType,
+ nullable: Boolean): Schema = {
+ (nullable, avroType.getType) match {
+ case (false, Type.UNION) | (true, Type.UNION) =>
+ // avro uses union to represent nullable type.
+ val fieldTypes = avroType.getTypes.asScala
+
+ // If we're nullable, we need to have at least two types. Cases with more than two types
+ // are captured in test("read read-write, read-write w/ schema, read") w/ test.avro input
+ assert(fieldTypes.length >= 2)
+
+ val actualType = catalystType match {
+ case NullType => fieldTypes.filter(_.getType == Type.NULL)
+ case BooleanType => fieldTypes.filter(_.getType == Type.BOOLEAN)
+ case ByteType => fieldTypes.filter(_.getType == Type.INT)
+ case BinaryType =>
+ val at = fieldTypes.filter(x => x.getType == Type.BYTES || x.getType == Type.FIXED)
+ if (at.length > 1) {
+ throw new IncompatibleSchemaException(
+ s"Cannot resolve schema of ${catalystType} against union ${avroType.toString}")
+ } else {
+ at
+ }
+ case ShortType | IntegerType => fieldTypes.filter(_.getType == Type.INT)
+ case LongType => fieldTypes.filter(_.getType == Type.LONG)
+ case FloatType => fieldTypes.filter(_.getType == Type.FLOAT)
+ case DoubleType => fieldTypes.filter(_.getType == Type.DOUBLE)
+ case d: DecimalType => fieldTypes.filter(_.getType == Type.STRING)
+ case StringType => fieldTypes
+ .filter(x => x.getType == Type.STRING || x.getType == Type.ENUM)
+ case DateType => fieldTypes.filter(x => x.getType == Type.INT || x.getType == Type.LONG)
+ case TimestampType => fieldTypes.filter(_.getType == Type.LONG)
+ case ArrayType(et, containsNull) =>
+ // Find array that matches the element type specified
+ fieldTypes.filter(x => x.getType == Type.ARRAY
+ && typeMatchesSchema(et, x.getElementType))
+ case st: StructType => // Find the matching record!
+ val recordTypes = fieldTypes.filter(x => x.getType == Type.RECORD)
+ if (recordTypes.length > 1) {
+ throw new IncompatibleSchemaException(
+ "Unions of multiple record types are NOT supported with user-specified schema")
+ }
+ recordTypes
+ case MapType(kt, vt, valueContainsNull) =>
+ // Find the map that matches the value type. Maps in Avro are always key type string
+ fieldTypes.filter(x => x.getType == Type.MAP && typeMatchesSchema(vt, x.getValueType))
+ case other =>
+ throw new IncompatibleSchemaException(s"Unexpected type: $other")
+ }
+
+ assert(actualType.length == 1)
--- End diff --
I've removed ALL asserts from the code and replaced them with clearer exceptions.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #93707 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93707/testReport)** for PR 21847 at commit [`874d2a8`](https://github.com/apache/spark/commit/874d2a86369083ba271e737b36f19baf2710ee01).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL][WIP]: Built-in AVRO support...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Merged build finished. Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Build finished. Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL][WIP]: Built-in AVRO support...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #93604 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93604/testReport)** for PR 21847 at commit [`6f686d6`](https://github.com/apache/spark/commit/6f686d6401fa991f6bcb0613bd9c4256cc7cffb9).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by lindblombr <gi...@git.apache.org>.
Github user lindblombr commented on the issue:
https://github.com/apache/spark/pull/21847
FYI Doing some additional testing around performance and have found a pretty gnarly regression with a particular type of schema. I'll try to track down what's causing it.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL][WIP]: Built-in AVRO support...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/93457/
Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by lindblombr <gi...@git.apache.org>.
Github user lindblombr commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r205912386
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -165,16 +183,112 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
result
}
- private def resolveNullableType(avroType: Schema, nullable: Boolean): Schema = {
- if (nullable) {
- // avro uses union to represent nullable type.
- val fields = avroType.getTypes.asScala
- assert(fields.length == 2)
- val actualType = fields.filter(_.getType != NULL)
- assert(actualType.length == 1)
- actualType.head
+ // Resolve an Avro union against a supplied DataType, i.e. a LongType compared against
+ // a ["null", "long"] should return a schema of type Schema.Type.LONG
+ // This function also handles resolving a DataType against unions of 2 or more types, i.e.
+ // an IntType resolves against a ["int", "long", "null"] will correctly return a schema of
+ // type Schema.Type.LONG
+ private def resolveNullableType(avroType: Schema, catalystType: DataType,
+ nullable: Boolean): Schema = {
+ (nullable, avroType.getType) match {
--- End diff --
If-else is done. Much cleaner.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL][WIP]: Built-in AVRO support...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/93466/
Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #93699 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93699/testReport)** for PR 21847 at commit [`1671ade`](https://github.com/apache/spark/commit/1671adeac3d306372f213794e93396c2bc57776f).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/93728/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL][WIP]: Built-in AVRO support...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #93466 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93466/testReport)** for PR 21847 at commit [`033f4dd`](https://github.com/apache/spark/commit/033f4dd11b068072f023312b8cd5ae52fef769bc).
* This patch **fails Scala style tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by dbtsai <gi...@git.apache.org>.
Github user dbtsai commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r206353416
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -120,7 +133,7 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
case MapType(kt, vt, valueContainsNull) if kt == StringType =>
val valueConverter = newConverter(
- vt, resolveNullableType(avroType.getValueType, valueContainsNull))
+ vt, resolveUnionType(avroType.getValueType, vt, valueContainsNull))
--- End diff --
ditto for MapType,
```scala
(getter, ordinal) =>
val mapData = getter.getMap(ordinal)
val length = mapData.numElements()
val result = new java.util.HashMap[String, Any](length)
val keyArray = mapData.keyArray()
val valueArray = mapData.valueArray()
var i = 0
while (i < length) {
val key = keyArray.getUTF8String(i).toString
if (valueContainsNull && valueArray.isNullAt(i)) {
result.put(key, null)
} else {
result.put(key, valueConverter(valueArray, i))
}
i += 1
}
result
```
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #94531 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/94531/testReport)** for PR 21847 at commit [`df486f2`](https://github.com/apache/spark/commit/df486f23d763450800bc7d8c4f03224d6fad1df0).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL][WIP]: Built-in AVRO support...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/93464/
Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by lindblombr <gi...@git.apache.org>.
Github user lindblombr commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r206746980
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -165,16 +182,118 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
result
}
- private def resolveNullableType(avroType: Schema, nullable: Boolean): Schema = {
- if (nullable) {
+ // Resolve an Avro union against a supplied DataType, i.e. a LongType compared against
+ // a ["null", "long"] should return a schema of type Schema.Type.LONG
+ // This function also handles resolving a DataType against unions of 2 or more types, i.e.
+ // an IntType resolves against a ["int", "long", "null"] will correctly return a schema of
+ // type Schema.Type.LONG
+ private def resolveUnionType(avroType: Schema, catalystType: DataType,
+ nullable: Boolean): Schema = {
+ if (avroType.getType == Type.UNION) {
// avro uses union to represent nullable type.
- val fields = avroType.getTypes.asScala
- assert(fields.length == 2)
- val actualType = fields.filter(_.getType != NULL)
- assert(actualType.length == 1)
+ val fieldTypes = avroType.getTypes.asScala
+
+ // If we're nullable, we need to have at least two types. Cases with more than two types
+ // are captured in test("read read-write, read-write w/ schema, read") w/ test.avro input
+ if (nullable && fieldTypes.length < 2) {
+ throw new IncompatibleSchemaException(
+ s"Cannot resolve nullable ${catalystType} against union type ${avroType}")
+ }
+
+ val actualType = catalystType match {
+ case NullType => fieldTypes.filter(_.getType == Type.NULL)
+ case BooleanType => fieldTypes.filter(_.getType == Type.BOOLEAN)
+ case ByteType => fieldTypes.filter(_.getType == Type.INT)
+ case BinaryType =>
+ val at = fieldTypes.filter(x => x.getType == Type.BYTES || x.getType == Type.FIXED)
+ if (at.length > 1) {
+ throw new IncompatibleSchemaException(
+ s"Cannot resolve schema of ${catalystType} against union ${avroType.toString}")
+ } else {
+ at
+ }
+ case ShortType | IntegerType => fieldTypes.filter(_.getType == Type.INT)
+ case LongType => fieldTypes.filter(_.getType == Type.LONG)
+ case FloatType => fieldTypes.filter(_.getType == Type.FLOAT)
+ case DoubleType => fieldTypes.filter(_.getType == Type.DOUBLE)
+ case d: DecimalType => fieldTypes.filter(_.getType == Type.STRING)
+ case StringType => fieldTypes
+ .filter(x => x.getType == Type.STRING || x.getType == Type.ENUM)
+ case DateType => fieldTypes.filter(x => x.getType == Type.INT || x.getType == Type.LONG)
+ case TimestampType => fieldTypes.filter(_.getType == Type.LONG)
+ case ArrayType(et, containsNull) =>
+ // Find array that matches the element type specified
+ fieldTypes.filter(x => x.getType == Type.ARRAY
+ && typeMatchesSchema(et, x.getElementType))
+ case st: StructType => // Find the matching record!
+ val recordTypes = fieldTypes.filter(x => x.getType == Type.RECORD)
+ if (recordTypes.length > 1) {
+ throw new IncompatibleSchemaException(
+ "Unions of multiple record types are NOT supported with user-specified schema")
+ }
+ recordTypes
+ case MapType(kt, vt, valueContainsNull) =>
+ // Find the map that matches the value type. Maps in Avro are always key type string
+ fieldTypes.filter(x => x.getType == Type.MAP && typeMatchesSchema(vt, x.getValueType))
--- End diff --
In `SchemaConverters.toAvro`, the expectation is that Maps are keyed only with `StringType`:
case MapType(StringType, vt, valueContainsNull) =>
builder.map().values(toAvroType(vt, valueContainsNull, recordName, prevNameSpace))
When you attempt this trivial test case, we fail
```
test("SPARK-24855: Maps with kv not string") {
withTempPath { dir =>
val someData = Seq(
Row("a", Map(
1 -> "foo",
2 -> "bar",
3 -> "baz"
)
),
Row("b", Map(
1 -> "foo",
2 -> "bar",
3 -> "baz"
)
)
)
val someSchema = StructType(Seq(
StructField("id", StringType, true),
StructField("map", MapType(IntegerType, StringType), true)
)
)
val df = spark.createDataFrame(
spark.sparkContext.parallelize(someData), someSchema
)
df.write.format("avro").save("dataset")
}
```
Exception as follows
```
Unexpected type MapType(IntegerType,StringType,true).
org.apache.spark.sql.avro.IncompatibleSchemaException: Unexpected type MapType(IntegerType,StringType,true).
at org.apache.spark.sql.avro.SchemaConverters$.toAvroType(SchemaConverters.scala:136)
at org.apache.spark.sql.avro.SchemaConverters$$anonfun$toAvroType$1.apply(SchemaConverters.scala:130)
at org.apache.spark.sql.avro.SchemaConverters$$anonfun$toAvroType$1.apply(SchemaConverters.scala:129)
```
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL][WIP]: Built-in AVRO support...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #93603 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93603/testReport)** for PR 21847 at commit [`be037bd`](https://github.com/apache/spark/commit/be037bda9dbb1eee267ccf5ff1391d52cae93714).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/93837/
Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO support shou...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/21847
**[Test build #94522 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/94522/testReport)** for PR 21847 at commit [`03e112c`](https://github.com/apache/spark/commit/03e112c10c7a5f5b96c192f1388a96c74ff162f4).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #21847: [SPARK-24855][SQL][EXTERNAL]: Built-in AVRO suppo...
Posted by dbtsai <gi...@git.apache.org>.
Github user dbtsai commented on a diff in the pull request:
https://github.com/apache/spark/pull/21847#discussion_r209042005
--- Diff: external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala ---
@@ -87,10 +87,18 @@ class AvroSerializer(rootCatalystType: DataType, rootAvroType: Schema, nullable:
(getter, ordinal) => getter.getDouble(ordinal)
case d: DecimalType =>
(getter, ordinal) => getter.getDecimal(ordinal, d.precision, d.scale).toString
- case StringType =>
- (getter, ordinal) => new Utf8(getter.getUTF8String(ordinal).getBytes)
- case BinaryType =>
- (getter, ordinal) => ByteBuffer.wrap(getter.getBinary(ordinal))
+ case StringType => avroType.getType match {
+ case Type.ENUM =>
+ (getter, ordinal) => new EnumSymbol(avroType, getter.getUTF8String(ordinal).toString)
+ case _ =>
+ (getter, ordinal) => new Utf8(getter.getUTF8String(ordinal).getBytes)
+ }
+ case BinaryType => avroType.getType match {
+ case Type.FIXED =>
--- End diff --
I think we should throw an exception if the size of the bytes to be written is different from the "size" attribute without doing the padding thing.
BTW, avro doesn't handle this well. If the size to be written is larger than the "size" attribute, avro will silently cause data corruption.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL][WIP]: Built-in AVRO support...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/93606/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #21847: [SPARK-24855][SQL][EXTERNAL][WIP]: Built-in AVRO support...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/21847
Merged build finished. Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org