[SPARK-25213][PYTHON] Add project to v2 scans before python filters. #22206
Conversation
This comment has been minimized.
This comment has been minimized.
This comment has been minimized.
This comment has been minimized.
|
Test build #95173 has finished for PR 22206 at commit
|
| ProjectExec(project, withFilter) :: Nil | ||
| if (project.exists(hasScalarPythonUDF)) { | ||
| val references = project.map(_.references).reduce(_ ++ _).toSeq | ||
| ProjectExec(project, ProjectExec(references, withFilter)) :: Nil |
There was a problem hiding this comment.
Why do we need to add extra Project on top of Filter here?
There was a problem hiding this comment.
The v2 data sources return InternalRow, not UnsafeRow. Python UDFs can't handle InternalRow, so this is intended to add a projection to convert to unsafe before the projection that contains a python UDF.
There was a problem hiding this comment.
oh, I see. It is also used to make sure PythonUDF in top Project takes unsafe row input.
There was a problem hiding this comment.
nit: If we already add Project on top of Filter, we don't need to add another Project here, right?
There was a problem hiding this comment.
That one was only added if there was a filter and if that filter ran a UDF. This will add an unnecessary project if both the filter and the project have python UDFs, but I thought that was probably okay. I can add a boolean to signal if the filter caused one to be added already if you think it's worth it.
| from pyspark.sql.functions import udf | ||
|
|
||
| df = self.spark.read.format("org.apache.spark.sql.sources.v2.SimpleDataSourceV2").load() | ||
| result = df.withColumn('x', udf(lambda x: x, 'int')(df['i'])) |
There was a problem hiding this comment.
This only tests Project with Scalar PythonUDF? Might be better to also test Filter case.
There was a problem hiding this comment.
Agreed. I was just verifying that the fix worked before spending more time on it.
|
Test build #95187 has finished for PR 22206 at commit
|
| def test_pyspark_udf_SPARK_25213(self): | ||
| from pyspark.sql.functions import udf | ||
|
|
||
| df = self.spark.read.format("org.apache.spark.sql.sources.v2.SimpleDataSourceV2").load() |
There was a problem hiding this comment.
I think this test will fail if test classes are not compiled. Can we check if test classes are compiled and then skip if not existent?
|
|
||
|
|
||
| class DataSourceV2Tests(ReusedSQLTestCase): | ||
| def test_pyspark_udf_SPARK_25213(self): |
There was a problem hiding this comment.
not a big deal but I would avoid SPARK_25213 postfix at the end just for consistency.
There was a problem hiding this comment.
I like that the tests in Scala include this information somewhere. Is there a better place for it in PySpark? I'm not aware of another way to pass extra metadata, but I'm open to if it there's a better way.
|
Not a big deal but PR title: |
rdblue
changed the title
|
@HyukjinKwon and @viirya, thank you for looking at this commit, but I like @cloud-fan's approach to fixing this in #22244 better than this work-around. I'm going to close this in favor of that approach, although if we need a quick fix I can pick this back up. |
4 participants
What changes were proposed in this pull request?
The v2 API always adds a projection when converting to physical plan to ensure that rows are all
UnsafeRow. This is added after any filters run by Spark, assuming that the filter and projection can handle InternalRow, but this fails if those nodes contain python UDFs. This PR detects the Python UDFs and adds a projection above the filter to immediately convert toUnsafeRowbefore passing data to python.How was this patch tested?
This adds a test for the case reported in SPARK-25213 in python's SQL tests.