[SPARK-11743][SQL] Add UserDefinedType support to RowEncoder

[viirya]

JIRA: https://issues.apache.org/jira/browse/SPARK-11743

RowEncoder doesn't support UserDefinedType now. We should add the support for it.

[SparkQA]

Test build #45923 has finished for PR 9712 at commit 2721741.

• This patch fails Scala style tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #45926 has finished for PR 9712 at commit bab5c5b.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[cloud-fan]

I have a question here. According to the Javadoc of Row, a user should use and only use Row for StructType field. It's ok to support Product too, but do we have a reason for this? Is it needed for the UDT stuff? Sorry I'm not familiar with UDT handling, it will be good if you can explain it in detail, thanks!

[viirya]

Actually I found this problem when working on ScalaUDF. ScalaUDF will use schemaFor to obtain catalyst type for UDF input and output. The catalyst type returned by schemaFor for a Product is StructType. It is reasonable as we don't have other type to represent Product as I see.

So for a StructType field in an external Row, both Row and Product are possible values. When we call extractorsFor on the external Row, externalDataTypeFor will return ObjectType(classOf[Row]) for this field. But the get accessor on the inputObject (i.e., the Row) will possibly return a Product for the ScalaUDF case and an exception will be thrown.

[cloud-fan]

If one of the input parameter is Tuple2, then we need to use the encoder to decode a catalyst value to external value, i.e. decode an InternalRow object to Tuple2 object. I think this is hard for a RowEncoder(your change only makes it possible to encode a Product into InternalRow, but not vice versa), we should use ProductEncoder for this case.

[viirya]

If we have an input parameter mapping to a StructType field in an InternalRow, we will use Row as its input type. E.g., sqlContext.udf.register("udfFunc", (ns: Row) => { (ns.getInt(0), ns.getString(1)) }). But we can't use Row as output type for an UDF. Because we can still get the input schema of ScalaUDF's children expressions later if we can't infer input types correctly by using schemaFor. However, the output types of the UDF can be only inferred by schemaFor.

[cloud-fan]

Ah I see. We need the type tag to infer the return type of UDF, and if a Row is returned, there is no type information we can get. How about we use ProductEncoder or FlatEncoder for the return value?

[viirya]

Hmm, as I tried, I found a problem is we may not always be able to get the type T needed to construct ProductEncoder and FlatEncoder. Even we can get it, we can't keep it in ScalaUDF due to serialization issue. So I think using RowEncoder is more reasonable.

[davies]

I think the reason we still support Product for StructType is for backward-compatibility, we did not enforce the inbound type before, someone may reply one it (because it's easier than Row in Scala).

[SparkQA]

Test build #45931 has finished for PR 9712 at commit db644fb.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[davies]

Should we merge these branches together?

[viirya]

yes. updated.

[SparkQA]

Test build #45992 has finished for PR 9712 at commit a5fdbce.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):\n * case class JSONOptions(\n

[viirya]

retest this please.

[SparkQA]

Test build #45994 has finished for PR 9712 at commit a5fdbce.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):\n * case class JSONOptions(\n

[davies]

LGTM, merging this into master and 1.6 branch, thanks!

[cloud-fan]

any reason why we put this test here instead of adding arrayOfUDT type in encodeDecodeTest like structOfUDT?

[viirya]

No. I moved it in #10538.