{"action":"created","issue":{"url":"https://api.github.com/repos/apache/spark/issues/16193","repository_url":"https://api.github.com/repos/apache/spark","labels_url":"https://api.github.com/repos/apache/spark/issues/16193/labels{/name}","comments_url":"https://api.github.com/repos/apache/spark/issues/16193/comments","events_url":"https://api.github.com/repos/apache/spark/issues/16193/events","html_url":"https://github.com/apache/spark/pull/16193","id":193985187,"number":16193,"title":"[SPARK-18766] [SQL] Push Down Filter Through BatchEvalPython (Python UDF)","user":{"login":"gatorsmile","id":11567269,"avatar_url":"https://avatars.githubusercontent.com/u/11567269?v=3","gravatar_id":"","url":"https://api.github.com/users/gatorsmile","html_url":"https://github.com/gatorsmile","followers_url":"https://api.github.com/users/gatorsmile/followers","following_url":"https://api.github.com/users/gatorsmile/following{/other_user}","gists_url":"https://api.github.com/users/gatorsmile/gists{/gist_id}","starred_url":"https://api.github.com/users/gatorsmile/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/gatorsmile/subscriptions","organizations_url":"https://api.github.com/users/gatorsmile/orgs","repos_url":"https://api.github.com/users/gatorsmile/repos","events_url":"https://api.github.com/users/gatorsmile/events{/privacy}","received_events_url":"https://api.github.com/users/gatorsmile/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":9,"created_at":"2016-12-07T08:20:48Z","updated_at":"2016-12-07T21:33:26Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/16193","html_url":"https://github.com/apache/spark/pull/16193","diff_url":"https://github.com/apache/spark/pull/16193.diff","patch_url":"https://github.com/apache/spark/pull/16193.patch"},"body":"### What changes were proposed in this pull request?\r\nCurrently, when users use Python UDF in Filter, BatchEvalPython is always generated below FilterExec. However, not all the predicates need to be evaluated after Python UDF execution. Thus, this PR is to push down the determinisitc predicates through `BatchEvalPython`.\r\n```\r\n>>> df = spark.createDataFrame([(1, \"1\"), (2, \"2\"), (1, \"2\"), (1, \"2\")], [\"key\", \"value\"])\r\n>>> from pyspark.sql.functions import udf, col\r\n>>> from pyspark.sql.types import BooleanType\r\n>>> my_filter = udf(lambda a: a < 2, BooleanType())\r\n>>> sel = df.select(col(\"key\"), col(\"value\")).filter((my_filter(col(\"key\"))) & (df.value < \"2\"))\r\n>>> sel.explain(True)\r\n```\r\nBefore the fix, the plan looks like\r\n```\r\n== Optimized Logical Plan ==\r\nFilter ((isnotnull(value#1) && <lambda>(key#0L)) && (value#1 < 2))\r\n+- LogicalRDD [key#0L, value#1]\r\n\r\n== Physical Plan ==\r\n*Project [key#0L, value#1]\r\n+- *Filter ((isnotnull(value#1) && pythonUDF0#9) && (value#1 < 2))\r\n   +- BatchEvalPython [<lambda>(key#0L)], [key#0L, value#1, pythonUDF0#9]\r\n      +- Scan ExistingRDD[key#0L,value#1]\r\n```\r\n\r\nAfter the fix, the plan looks like\r\n```\r\n== Optimized Logical Plan ==\r\nFilter ((isnotnull(value#1) && <lambda>(key#0L)) && (value#1 < 2))\r\n+- LogicalRDD [key#0L, value#1]\r\n\r\n== Physical Plan ==\r\n*Project [key#0L, value#1]\r\n+- *Filter pythonUDF0#9: boolean\r\n   +- BatchEvalPython [<lambda>(key#0L)], [key#0L, value#1, pythonUDF0#9]\r\n      +- *Filter (isnotnull(value#1) && (value#1 < 2))\r\n         +- Scan ExistingRDD[key#0L,value#1]\r\n```\r\n\r\n### How was this patch tested?\r\nAdded both unit test cases for `BatchEvalPythonExec` and also add an end-to-end test case in Python test suite. "},"comment":{"url":"https://api.github.com/repos/apache/spark/issues/comments/265580076","html_url":"https://github.com/apache/spark/pull/16193#issuecomment-265580076","issue_url":"https://api.github.com/repos/apache/spark/issues/16193","id":265580076,"user":{"login":"AmplabJenkins","id":4031291,"avatar_url":"https://avatars.githubusercontent.com/u/4031291?v=3","gravatar_id":"","url":"https://api.github.com/users/AmplabJenkins","html_url":"https://github.com/AmplabJenkins","followers_url":"https://api.github.com/users/AmplabJenkins/followers","following_url":"https://api.github.com/users/AmplabJenkins/following{/other_user}","gists_url":"https://api.github.com/users/AmplabJenkins/gists{/gist_id}","starred_url":"https://api.github.com/users/AmplabJenkins/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/AmplabJenkins/subscriptions","organizations_url":"https://api.github.com/users/AmplabJenkins/orgs","repos_url":"https://api.github.com/users/AmplabJenkins/repos","events_url":"https://api.github.com/users/AmplabJenkins/events{/privacy}","received_events_url":"https://api.github.com/users/AmplabJenkins/received_events","type":"User","site_admin":false},"created_at":"2016-12-07T21:33:26Z","updated_at":"2016-12-07T21:33:26Z","body":"Test FAILed.\nRefer to this link for build results (access rights to CI server needed): \nhttps://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/69808/\nTest FAILed."}}
{"actor":{"display_login":"AmplabJenkins"}}