Skip to content

[SPARK-12439][SQL] Fix toCatalystArray and MapObjects #10391

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
viirya wants to merge 10 commits into from
Closed

[SPARK-12439][SQL] Fix toCatalystArray and MapObjects #10391

Show file tree
Hide file tree
Changes from 8 commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
788a7c6
Fix toCatalystArray and MapObjects.
viirya Dec 19, 2015
180048c
Move test case.
viirya Dec 22, 2015
a726fd8
Remove unnecessary classes.
viirya Dec 22, 2015
090b293
Fix MapObjects.
viirya Dec 23, 2015
7389e15
Remove unnecessary import.
viirya Dec 23, 2015
ae55e01
Add check if genFunction returns an Option.
viirya Dec 23, 2015
f5b9d50
Check WrapOption.
viirya Dec 23, 2015
7728124
Do not check WrapOption. Instead, we should wrap an If expression out…
viirya Dec 29, 2015
e0d34a5
Merge remote-tracking branch 'upstream/master' into fix-catalystarray
viirya Dec 31, 2015
e541e04
Rebase updates.
viirya Dec 31, 2015
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
13 changes: 8 additions & 5 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/ScalaReflection.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -406,11 +406,14 @@ object ScalaReflection extends ScalaReflection {
def toCatalystArray(input: Expression, elementType: `Type`): Expression = {
val externalDataType = dataTypeFor(elementType)
val Schema(catalystType, nullable) = silentSchemaFor(elementType)
if (isNativeType(catalystType)) {
NewInstance(
classOf[GenericArrayData],
input :: Nil,
dataType = ArrayType(catalystType, nullable))
if (isNativeType(externalDataType)) {
expressions.If(
IsNull(input),
expressions.Literal.create(null, ArrayType(catalystType, nullable)),
NewInstance(
classOf[GenericArrayData],
input :: Nil,
dataType = ArrayType(catalystType, nullable)))
Copy link
Member Author

@viirya viirya Dec 23, 2015

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The part of fixing is from pr #10401. Without this, the following fixing to MapObjects will cause null exception. Can we merge #10401 first then I do rebase here?

Copy link
Member Author

@viirya viirya Dec 29, 2015

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Waiting #10443 to getting merged too.

} else {
val clsName = getClassNameFromType(elementType)
val newPath = s"""- array element class: "$clsName"""" +: walkedTypePath
Expand Down
13 changes: 10 additions & 3 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 @@ -35,7 +35,8 @@ object RowEncoder {
def apply(schema: StructType): ExpressionEncoder[Row] = {
val cls = classOf[Row]
val inputObject = BoundReference(0, ObjectType(cls), nullable = true)
val extractExpressions = extractorsFor(inputObject, schema)
// We use an If expression to wrap extractorsFor result of StructType
val extractExpressions = extractorsFor(inputObject, schema).asInstanceOf[If].falseValue
val constructExpression = constructorFor(schema)
Copy link
Contributor

@cloud-fan cloud-fan Dec 29, 2015

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

same with extractExpressions, we should also strip the top-most If for constructExpression

Copy link
Member Author

@viirya viirya Dec 29, 2015

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Unlike extractExpressions, constructExpression hasn't been wrapped with an If here. Because there are two constructorFor methods, and the one wrapped with If is the second one.

new ExpressionEncoder[Row](
schema,
Expand Down Expand Up @@ -92,6 +93,7 @@ object RowEncoder {
NewInstance(
classOf[GenericArrayData],
inputObject :: Nil,
propagateNull = true,
Copy link
Member Author

@viirya viirya Dec 29, 2015

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

#10443 will fix this by using true as default value for propagateNull. As that pr is not merged yet, in order to pass tests now, I set propagateNull as true here. Once #10443 is getting merged, I can remove the two lines here.

dataType = t)
case _ => MapObjects(extractorsFor(_, et), inputObject, externalDataTypeFor(et))
}
Expand All @@ -114,6 +116,7 @@ object RowEncoder {
NewInstance(
classOf[ArrayBasedMapData],
convertedKeys :: convertedValues :: Nil,
propagateNull = true,
dataType = t)

case StructType(fields) =>
Expand All @@ -130,7 +133,9 @@ object RowEncoder {
Invoke(inputObject, method, externalDataTypeFor(f.dataType), Literal(i) :: Nil),
f.dataType))
}
CreateStruct(convertedFields)
If(IsNull(inputObject),
Literal.create(null, inputType),
CreateStruct(convertedFields))
}

private def externalDataTypeFor(dt: DataType): DataType = dt match {
Expand Down Expand Up @@ -222,6 +227,8 @@ object RowEncoder {
Literal.create(null, externalDataTypeFor(f.dataType)),
constructorFor(GetStructField(input, i)))
}
CreateExternalRow(convertedFields)
If(IsNull(input),
Literal.create(null, externalDataTypeFor(input.dataType)),
CreateExternalRow(convertedFields))
}
}
4 changes: 2 additions & 2 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/objects.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -456,10 +456,10 @@ case class MapObjects(
($elementJavaType)${genInputData.value}${itemAccessor(loopIndex)};
$loopNullCheck

if (${loopVar.isNull}) {
${genFunction.code}
if (${genFunction.isNull}) {
Copy link
Member Author

@viirya viirya Dec 23, 2015

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Even loopVar.isNull is true, we should still call genFunction because it may have special handling for null case, for example WrapOption will produce a None for null value. We should depend on genFunction.isNull to decide $convertedArray[$loopIndex] is null or not.

$convertedArray[$loopIndex] = null;
} else {
${genFunction.code}
$convertedArray[$loopIndex] = ${genFunction.value};
}

Expand Down
3 changes: 3 additions & 0 deletions ...talyst/src/test/scala/org/apache/spark/sql/catalyst/encoders/ExpressionEncoderSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -155,6 +155,9 @@ class ExpressionEncoderSuite extends SparkFunSuite {

productTest(OptionalData(None, None, None, None, None, None, None, None))

encodeDecodeTest(Seq(Some(1), None), "Option in array")
encodeDecodeTest(Map(1 -> Some(10L), 2 -> Some(20L), 3 -> None), "Option in map")

productTest(BoxedData(1, 1L, 1.0, 1.0f, 1.toShort, 1.toByte, true))

productTest(BoxedData(null, null, null, null, null, null, null))
Expand Down