Skip to content

[SPARK-16167][SQL] RowEncoder should preserve array/map type nullability. #13873

New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

Closed
wants to merge 8 commits into from
Closed

[SPARK-16167][SQL] RowEncoder should preserve array/map type nullability. #13873

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
f64f570
Add tests to check if RowEncoder preserves array/map nullability.
ueshin Jun 23, 2016
093a9fa
Modify RowEncoder and MapObjects to preserve array/map nullability.
ueshin Jun 23, 2016
be6ed48
Fix nullability of NewInstance.
ueshin Jun 23, 2016
f30d7d3
Merge branch 'master' into issues/SPARK-16167
ueshin Aug 18, 2016
b22867b
Merge branch 'master' into issues/SPARK-16167
ueshin Aug 31, 2016
306b283
Merge branch 'master' into issues/SPARK-16167
ueshin Jun 29, 2017
1cc58a6
Merge branch 'master' into issues/SPARK-16167
ueshin Jul 5, 2017
9caf425
Address a comment.
ueshin Jul 5, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 21 additions & 4 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/encoders/RowEncoder.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -123,7 +123,7 @@ object RowEncoder {
inputObject :: Nil,
returnNullable = false)

case t @ ArrayType(et, cn) =>
case t @ ArrayType(et, containsNull) =>
et match {
case BooleanType | ByteType | ShortType | IntegerType | LongType | FloatType | DoubleType =>
StaticInvoke(
Expand All @@ -132,8 +132,16 @@ object RowEncoder {
"toArrayData",
inputObject :: Nil,
returnNullable = false)

case _ => MapObjects(
element => serializerFor(ValidateExternalType(element, et), et),
element => {
val value = serializerFor(ValidateExternalType(element, et), et)
if (!containsNull) {
AssertNotNull(value, Seq.empty)
} else {
value
}
},
inputObject,
ObjectType(classOf[Object]))
}
Expand All @@ -155,10 +163,19 @@ object RowEncoder {
ObjectType(classOf[scala.collection.Seq[_]]), returnNullable = false)
val convertedValues = serializerFor(values, ArrayType(vt, valueNullable))

NewInstance(
val nonNullOutput = NewInstance(
classOf[ArrayBasedMapData],
convertedKeys :: convertedValues :: Nil,
dataType = t)
dataType = t,
propagateNull = false)

if (inputObject.nullable) {
If(IsNull(inputObject),
Literal.create(null, inputType),
nonNullOutput)
} else {
nonNullOutput
}

case StructType(fields) =>
val nonNullOutput = CreateNamedStruct(fields.zipWithIndex.flatMap { case (field, index) =>
Expand Down
33 changes: 33 additions & 0 deletions sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/encoders/RowEncoderSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -273,6 +273,39 @@ class RowEncoderSuite extends SparkFunSuite {
assert(e4.getMessage.contains("java.lang.String is not a valid external type"))
}

for {
elementType <- Seq(IntegerType, StringType)
containsNull <- Seq(true, false)
nullable <- Seq(true, false)
} {
test("RowEncoder should preserve array nullability: " +
s"ArrayType($elementType, containsNull = $containsNull), nullable = $nullable") {
val schema = new StructType().add("array", ArrayType(elementType, containsNull), nullable)
val encoder = RowEncoder(schema).resolveAndBind()
assert(encoder.serializer.length == 1)
assert(encoder.serializer.head.dataType == ArrayType(elementType, containsNull))
assert(encoder.serializer.head.nullable == nullable)
}
}

for {
keyType <- Seq(IntegerType, StringType)
valueType <- Seq(IntegerType, StringType)
valueContainsNull <- Seq(true, false)
nullable <- Seq(true, false)
} {
test("RowEncoder should preserve map nullability: " +
s"MapType($keyType, $valueType, valueContainsNull = $valueContainsNull), " +
s"nullable = $nullable") {
val schema = new StructType().add(
"map", MapType(keyType, valueType, valueContainsNull), nullable)
val encoder = RowEncoder(schema).resolveAndBind()
assert(encoder.serializer.length == 1)
assert(encoder.serializer.head.dataType == MapType(keyType, valueType, valueContainsNull))
assert(encoder.serializer.head.nullable == nullable)
}
}

private def encodeDecodeTest(schema: StructType): Unit = {
test(s"encode/decode: ${schema.simpleString}") {
val encoder = RowEncoder(schema).resolveAndBind()
Expand Down