[SPARK-14215] [SQL] [PYSPARK] Support chained Python UDFs

[davies]

What changes were proposed in this pull request?

This PR brings the support for chained Python UDFs, for example

select udf1(udf2(a))
select udf1(udf2(a) +3)
select udf1(udf2(a) + udf3(b)) 

Also directly chained unary Python UDFs are put in single batch of Python UDFs, others may require multiple batches.

For example,

>>>sqlContext.sql("select double(double(1))").explain()
==PhysicalPlan==WholeStageCodegen
: +-Project [pythonUDF#10 AS double(double(1))#9]
: +-INPUT+- !BatchPythonEvaluationdouble(double(1)), [pythonUDF#10]+-ScanOneRowRelation[]
>>>sqlContext.sql("select double(double(1) + double(2))").explain()
==PhysicalPlan==WholeStageCodegen
: +-Project [pythonUDF#19 AS double((double(1) + double(2)))#16]
: +-INPUT+- !BatchPythonEvaluationdouble((pythonUDF#17 + pythonUDF#18)), [pythonUDF#17,pythonUDF#18,pythonUDF#19]+- !BatchPythonEvaluationdouble(2), [pythonUDF#17,pythonUDF#18]+- !BatchPythonEvaluationdouble(1), [pythonUDF#17]+-ScanOneRowRelation[]

TODO: will support multiple unrelated Python UDFs in one batch (another PR).

How was this patch tested?

Added new unit tests for chained UDFs.

[davies]

cc @marmbrus@rxin

[SparkQA]

Test build #54364 has finished for PR 12014 at commit 024a822.

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

[davies]

cc @cloud-fan

[hvanhovell]

@davies I think the JIRA number should be SPARK-14215: https://issues.apache.org/jira/browse/SPARK-14215

[davies]

@hvanhovell Corrected, thanks!

[SparkQA]

Test build #54365 has finished for PR 12014 at commit b741073.

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

[cloud-fan]

what's the point of creating a new _wrap_function here? To decrease the size of serialized python function?

[cloud-fan]

oh i see, we wanna chain the functions at python worker side.

[SparkQA]

Test build #2705 has finished for PR 12014 at commit b741073.

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

[cloud-fan]

Overall LGTM

[viirya]

profiler seems need to be defined before this if block. The codes refer profiler later out of this block.

[davies]

The other branch also have profiler, so I think it's fine.

[SparkQA]

Test build #2707 has finished for PR 12014 at commit b741073.

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

[cloud-fan]

We should update the comments the explain our new strategy of extracting and evaluating python udfs.

[davies]

Updated.

[davies]

Merging this into master (the last commit only added comments).

[SparkQA]

Test build #54467 has finished for PR 12014 at commit c57e8a4.

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

[gatorsmile]

@davies It sounds you used a wrong JIRA number.

[davies]

@gatorsmile Corrected in PR, but the notification email is not updated.