[SPARK-9373][SQL] Support StructType in Tungsten projection

[rxin]

This pull request updates GenerateUnsafeProjection to support StructType. If an input struct type is backed already by an UnsafeRow, GenerateUnsafeProjection copies the bytes directly into its buffer space without any conversion. However, if the input is not an UnsafeRow, GenerateUnsafeProjection runs the code generated recursively to convert the input into an UnsafeRow and then copies it into the buffer space.

Also create a TungstenProject operator that projects data directly into UnsafeRow. Note that I'm not sure if this is the way we want to structure Unsafe+codegen operators, but we can defer that decision to follow-up pull requests.

[rxin]

there is a lot of code duplication between this one and createCode.

[SparkQA]

Test build #1206 has finished for PR 7689 at commit 0cae3e8.

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

[SparkQA]

Test build #38578 has finished for PR 7689 at commit f1bee3a.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • public static class StructWriter
  • public static class IntervalWriter
  • case class TungstenProject(projectList: Seq[NamedExpression], child: SparkPlan) extends UnaryNode

[SparkQA]

Test build #38590 has finished for PR 7689 at commit da6aed9.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • public static class StructWriter
  • case class CreateStructUnsafe(children: Seq[Expression]) extends Expression
  • case class CreateNamedStructUnsafe(children: Seq[Expression]) extends Expression
  • case class TungstenProject(projectList: Seq[NamedExpression], child: SparkPlan) extends UnaryNode

[rxin]

Jenkins, test this please.

[SparkQA]

Test build #38599 has finished for PR 7689 at commit f71b659.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • public static class StructWriter
  • case class CreateStructUnsafe(children: Seq[Expression]) extends Expression
  • case class CreateNamedStructUnsafe(children: Seq[Expression]) extends Expression
  • case class TungstenProject(projectList: Seq[NamedExpression], child: SparkPlan) extends UnaryNode

[SparkQA]

Test build #38598 timed out for PR 7689 at commit ce94678 after a configured wait of 175m.

[SparkQA]

Test build #38616 has finished for PR 7689 at commit be9f377.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • class RFormula(override val uid: String) extends Estimator[RFormulaModel] with RFormulaBase
  • public static class StructWriter
  • case class CreateStructUnsafe(children: Seq[Expression]) extends Expression
  • case class CreateNamedStructUnsafe(children: Seq[Expression]) extends Expression
  • case class TungstenProject(projectList: Seq[NamedExpression], child: SparkPlan) extends UnaryNode

[JoshRosen]

Looks like UnsafeRow.get doesn't support IntervalType:

 org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 1153.0 failed 1 times, most recent failure: Lost task 0.0 in stage 1153.0 (TID 4994, localhost): java.lang.UnsupportedOperationException: Unsupported data type interval
[info]      at org.apache.spark.sql.catalyst.expressions.UnsafeRow.get(UnsafeRow.java:271)
[info]      at org.apache.spark.sql.catalyst.CatalystTypeConverters$IdentityConverter.toScalaImpl(CatalystTypeConverters.scala:144)
[info]      at org.apache.spark.sql.catalyst.CatalystTypeConverters$CatalystTypeConverter.toScala(CatalystTypeConverters.scala:118)
[info]      at org.apache.spark.sql.catalyst.CatalystTypeConverters$StructConverter.toScala(CatalystTypeConverters.scala:268)
[info]      at org.apache.spark.sql.catalyst.CatalystTypeConverters$StructConverter.toScala(CatalystTypeConverters.scala:235)

[JoshRosen]

Typo: string -> struct

[JoshRosen]

This isn't obvious at first glance... since these expressions are returning names won't it be safe to just assume that they return strings and to use a .asInstanceOf[String] rather than calling toString here? Just a little wary of using .toString() during processing.

[JoshRosen]

Actually, I see that this is just copied from CreateNamedStruct above, so it should be fine.

[AmplabJenkins]

Merged build finished. Test FAILed.

[JoshRosen]

If you don't mind, can you also add a check to ensure that codegen is enabled?

[JoshRosen]

At a high level this seems fine to me, but I haven't looked in super-close detail yet. I'm fine with merging this pending tests and continuing to iterate in followup patches.

[rxin]

Thanks - going to merge this and submit a pr to fix the remaining issues tonight.