You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@spark.apache.org by "Junichi Koizumi (JIRA)" <ji...@apache.org> on 2019/04/27 12:28:00 UTC
[jira] [Issue Comment Deleted] (SPARK-27213) Unexpected results
when filter is used after distinct
[ https://issues.apache.org/jira/browse/SPARK-27213?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Junichi Koizumi updated SPARK-27213:
--------------------------------------
Comment: was deleted
(was: I have got the following output when returning the second query.
>>> df.filter("y_n='y'").select('x','y','z').distinct().explain(True)
== Parsed Logical Plan ==
Aggregate [x#6,y#7,z#8], [x#6,y#7,z#8]
+- Project [x#6,y#7,z#8]
+- Filter (y_n#9 = y)
+- LogicalRDD [x#6,y#7,z#8,y_n#9], MapPartitionsRDD[18] at applySchemaToPythonRDD at NativeMethodAccessorImpl.java:-2
== Analyzed Logical Plan ==
x: string, y: string, z: string
Aggregate [x#6,y#7,z#8], [x#6,y#7,z#8]
+- Project [x#6,y#7,z#8]
+- Filter (y_n#9 = y)
+- LogicalRDD [x#6,y#7,z#8,y_n#9], MapPartitionsRDD[18] at applySchemaToPythonRDD at NativeMethodAccessorImpl.java:-2
== Optimized Logical Plan ==
Aggregate [x#6,y#7,z#8], [x#6,y#7,z#8]
+- Project [x#6,y#7,z#8]
+- Filter (y_n#9 = y)
+- LogicalRDD [x#6,y#7,z#8,y_n#9], MapPartitionsRDD[18] at applySchemaToPythonRDD at NativeMethodAccessorImpl.java:-2
== Physical Plan ==
TungstenAggregate(key=[x#6,y#7,z#8], functions=[], output=[x#6,y#7,z#8])
+- TungstenExchange hashpartitioning(x#6,y#7,z#8,200), None
+- TungstenAggregate(key=[x#6,y#7,z#8], functions=[], output=[x#6,y#7,z#8])
+- Project [x#6,y#7,z#8]
+- Filter (y_n#9 = y)
+- Scan ExistingRDD[x#6,y#7,z#8,y_n#9]
)
> Unexpected results when filter is used after distinct
> -----------------------------------------------------
>
> Key: SPARK-27213
> URL: https://issues.apache.org/jira/browse/SPARK-27213
> Project: Spark
> Issue Type: Bug
> Components: PySpark
> Affects Versions: 2.3.2, 2.4.0
> Reporter: Rinaz Belhaj
> Priority: Major
> Labels: correctness, distinct, filter
>
> The following code gives unexpected output due to the filter not getting pushed down in catalyst optimizer.
> {code:java}
> df = spark.createDataFrame([['a','123','12.3','n'],['a','123','12.3','y'],['a','123','12.4','y']],['x','y','z','y_n'])
> df.show(5)
> df.filter("y_n='y'").select('x','y','z').distinct().show()
> df.select('x','y','z').distinct().filter("y_n='y'").show()
> {code}
> {panel:title=Output}
> |x|y|z|y_n|
> |a|123|12.3|n|
> |a|123|12.3|y|
> |a|123|12.4|y|
>
> |x|y|z|
> |a|123|12.3|
> |a|123|12.4|
>
> |x|y|z|
> |a|123|12.4|
> {panel}
> Ideally, the second statement should result in an error since the column used in the filter is not present in the preceding select statement. But the catalyst optimizer is using first() on column y_n and then applying the filter.
> Even if the filter was pushed down, the result would have been accurate.
> {code:java}
> df = spark.createDataFrame([['a','123','12.3','n'],['a','123','12.3','y'],['a','123','12.4','y']],['x','y','z','y_n'])
> df.filter("y_n='y'").select('x','y','z').distinct().explain(True)
> df.select('x','y','z').distinct().filter("y_n='y'").explain(True)
> {code}
> {panel:title=Output}
>
> == Parsed Logical Plan ==
> Deduplicate [x#74, y#75, z#76|#74, y#75, z#76]
> +- AnalysisBarrier
> +- Project [x#74, y#75, z#76|#74, y#75, z#76]
> +- Filter (y_n#77 = y)
> +- LogicalRDD [x#74, y#75, z#76, y_n#77|#74, y#75, z#76, y_n#77], false
>
> == Analyzed Logical Plan ==
> x: string, y: string, z: string
> Deduplicate [x#74, y#75, z#76|#74, y#75, z#76]
> +- Project [x#74, y#75, z#76|#74, y#75, z#76]
> +- Filter (y_n#77 = y)
> +- LogicalRDD [x#74, y#75, z#76, y_n#77|#74, y#75, z#76, y_n#77], false
>
> == Optimized Logical Plan ==
> Aggregate [x#74, y#75, z#76|#74, y#75, z#76], [x#74, y#75, z#76|#74, y#75, z#76]
> +- Project [x#74, y#75, z#76|#74, y#75, z#76]
> +- Filter (isnotnull(y_n#77) && (y_n#77 = y))
> +- LogicalRDD [x#74, y#75, z#76, y_n#77|#74, y#75, z#76, y_n#77], false
>
> == Physical Plan ==
> *(2) HashAggregate(keys=[x#74, y#75, z#76|#74, y#75, z#76], functions=[], output=[x#74, y#75, z#76|#74, y#75, z#76])
> +- Exchange hashpartitioning(x#74, y#75, z#76, 10)
> +- *(1) HashAggregate(keys=[x#74, y#75, z#76|#74, y#75, z#76], functions=[], output=[x#74, y#75, z#76|#74, y#75, z#76])
> +- *(1) Project [x#74, y#75, z#76|#74, y#75, z#76]
> +- *(1) Filter (isnotnull(y_n#77) && (y_n#77 = y))
> +- Scan ExistingRDD[x#74,y#75,z#76,y_n#77|#74,y#75,z#76,y_n#77]
>
> -------------------------------------------------------------------------------------------------------------------
>
> == Parsed Logical Plan ==
> 'Filter ('y_n = y)
> +- AnalysisBarrier
> +- Deduplicate [x#74, y#75, z#76|#74, y#75, z#76]
> +- Project [x#74, y#75, z#76|#74, y#75, z#76]
> +- LogicalRDD [x#74, y#75, z#76, y_n#77|#74, y#75, z#76, y_n#77], false
>
> == Analyzed Logical Plan ==
> x: string, y: string, z: string
> Project [x#74, y#75, z#76|#74, y#75, z#76]
> +- Filter (y_n#77 = y)
> +- Deduplicate [x#74, y#75, z#76|#74, y#75, z#76]
> +- Project [x#74, y#75, z#76, y_n#77|#74, y#75, z#76, y_n#77]
> +- LogicalRDD [x#74, y#75, z#76, y_n#77|#74, y#75, z#76, y_n#77], false
>
> == Optimized Logical Plan ==
> Project [x#74, y#75, z#76|#74, y#75, z#76]
> +- Filter (isnotnull(y_n#77) && (y_n#77 = y))
> +- Aggregate [x#74, y#75, z#76|#74, y#75, z#76], [x#74, y#75, z#76, first(y_n#77, false) AS y_n#77|#74, y#75, z#76, first(y_n#77, false) AS y_n#77]
> +- LogicalRDD [x#74, y#75, z#76, y_n#77|#74, y#75, z#76, y_n#77], false
>
> == Physical Plan ==
> *(3) Project [x#74, y#75, z#76|#74, y#75, z#76]
> +- *(3) Filter (isnotnull(y_n#77) && (y_n#77 = y))
> +- SortAggregate(key=[x#74, y#75, z#76|#74, y#75, z#76], functions=[first(y_n#77, false)|#77, false)], output=[x#74, y#75, z#76, y_n#77|#74, y#75, z#76, y_n#77])
> +- *(2) Sort [x#74 ASC NULLS FIRST, y#75 ASC NULLS FIRST, z#76 ASC NULLS FIRST|#74 ASC NULLS FIRST, y#75 ASC NULLS FIRST, z#76 ASC NULLS FIRST], false, 0
> +- Exchange hashpartitioning(x#74, y#75, z#76, 10)
> +- SortAggregate(key=[x#74, y#75, z#76|#74, y#75, z#76], functions=[partial_first(y_n#77, false)|#77, false)], output=[x#74, y#75, z#76, first#95, valueSet#96|#74, y#75, z#76, first#95, valueSet#96])
> +- *(1) Sort [x#74 ASC NULLS FIRST, y#75 ASC NULLS FIRST, z#76 ASC NULLS FIRST|#74 ASC NULLS FIRST, y#75 ASC NULLS FIRST, z#76 ASC NULLS FIRST], false, 0
> +- Scan ExistingRDD[x#74,y#75,z#76,y_n#77|#74,y#75,z#76,y_n#77]
>
> {panel}
> The second query. ie *"df.select('x','y','z').distinct().filter("y_n='y'").explain(True)"* should result in error rather than giving wrong output.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org