You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@spark.apache.org by "Apache Spark (Jira)" <ji...@apache.org> on 2022/05/11 04:55:00 UTC
[jira] [Assigned] (SPARK-39131) "Exists" is optimized too late (to LeftSemi) preventing filters to be inferred
[ https://issues.apache.org/jira/browse/SPARK-39131?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Apache Spark reassigned SPARK-39131:
------------------------------------
Assignee: Apache Spark
> "Exists" is optimized too late (to LeftSemi) preventing filters to be inferred
> ------------------------------------------------------------------------------
>
> Key: SPARK-39131
> URL: https://issues.apache.org/jira/browse/SPARK-39131
> Project: Spark
> Issue Type: Bug
> Components: Optimizer, SQL
> Affects Versions: 3.2.1
> Reporter: Alessandro Bellina
> Assignee: Apache Spark
> Priority: Major
>
> We would like to propose a slight change in the order of execution of logical plan optimizer rules given a performance issue we have seen with {{LeftSemi}} being materialized too late in the logical plan optimizer, and not benefiting from the null filtering that {{InferFiltersFromConstraints}} can insert.
> I have "something that works" locally (see rest of the description for info and a diff), but given that this is the optimizer it is not clear what else I could be breaking, so I'd like to hear from the experts on whether this is the right change.
> The query in question is based on TPCDS query16 which originally has an {{exists}} filter:
> {code:sql}
> …
> and exists (select *
> from catalog_sales cs2
> where cs1.cs_order_number = cs2.cs_order_number
> and cs1.cs_warehouse_sk <> cs2.cs_warehouse_sk)
> …
> {code}
> The rule {{RewritePredicateSubquery}} will turn this into a {{LeftSemi}} join like so:
> {code:sql}
> +- *(9) SortMergeJoin [cs_order_number#17L], [cs_order_number#872L], LeftSemi, NOT (cs_warehouse_sk#14 = cs_warehouse_sk#869)
> :- *(2) Sort [cs_order_number#17L ASC NULLS FIRST], false, 0
> : +- Exchange hashpartitioning(cs_order_number#17L, 200), ENSURE_REQUIREMENTS, [id=#364]
> : +- *(1) Filter ((isnotnull(cs_ship_date_sk#2) AND isnotnull(cs_ship_addr_sk#10)) AND isnotnull(cs_call_center_sk#11))
> : +- *(1) ColumnarToRow
> : +- FileScan parquet [...] Batched: true, DataFilters: [isnotnull(cs_ship_date_sk#2), isnotnull(cs_ship_addr_sk#10), isnotnull(cs_call_center_sk#11)],..., PushedFilters: [IsNotNull(cs_ship_date_sk), IsNotNull(cs_ship_addr_sk), IsNotNull(cs_call_center_sk)], ReadSchema: ...
> +- *(4) Sort [cs_order_number#872L ASC NULLS FIRST], false, 0
> +- Exchange hashpartitioning(cs_order_number#872L, 200), ENSURE_REQUIREMENTS, [id=#372]
> +- *(3) ColumnarToRow
> +- FileScan parquet [...] Batched: true, DataFilters: [], ..., PushedFilters: [], ReadSchema: ...
> {code}
> Note that the {{LeftSemi}} key and condition are not being null filtered on the stream side, and the build side has no filter at all. We have found that as the dataset size increases, this can become an issue, and in our case, it was many nulls that will not match. We would like to remove the unnecessary rows early at the scan and filter phases.
> The change we made allows the join key and the condition to be added to the stream side filter, and for the build side filter to get added:
> {code:sql}
> +- *(9) SortMergeJoin [cs_order_number#17L], [cs_order_number#943L], LeftSemi, NOT (cs_warehouse_sk#14 = cs_warehouse_sk#940)
> :- *(2) Sort [cs_order_number#17L ASC NULLS FIRST], false, 0
> : +- Exchange hashpartitioning(cs_order_number#17L, 200), ENSURE_REQUIREMENTS, [id=#759]
> : +-*(1) Filter ((((isnotnull(cs_ship_date_sk#2) AND isnotnull(cs_ship_addr_sk#10)) AND isnotnull(cs_call_center_sk#11)) AND isnotnull(cs_order_number#17L)) AND isnotnull(cs_warehouse_sk#14))
> : +- *(1) ColumnarToRow
> : +- FileScan parquet ..., DataFilters: [isnotnull(cs_ship_date_sk#2), isnotnull(cs_ship_addr_sk#10), isnotnull(cs_call_center_sk#11), is..., ..., PushedFilters: [IsNotNull(cs_ship_date_sk), IsNotNull(cs_ship_addr_sk), IsNotNull(cs_call_center_sk), IsNotNull(..., ReadSchema: ...
> +- *(4) Sort [cs_order_number#943L ASC NULLS FIRST], false, 0
> +- Exchange hashpartitioning(cs_order_number#943L, 200), ENSURE_REQUIREMENTS, [id=#768]
> +- *(3) Filter (isnotnull(cs_order_number#943L) AND isnotnull(cs_warehouse_sk#940))
> +- *(3) ColumnarToRow
> +- FileScan parquet ..., DataFilters: [isnotnull(cs_order_number#943L), isnotnull(cs_warehouse_sk#940)], ..., PartitionFilters: [], PushedFilters: [IsNotNull(cs_order_number), IsNotNull(cs_warehouse_sk)], ReadSchema: ...
> {code}
> This issue can be boiled down to this simple repro:
> {code:java}
> sc.parallelize((0 until 10).map(i => if (i%2 == 0) {null} else {Int.box(i)})).toDF.write.parquet("file:///tmp/my_test_table")
> spark.read.parquet("file:///tmp/my_test_table").createOrReplaceTempView("my_table")
> spark.sql("select * from my_table t1 where exists(select * from my_table t2 where t2.value = t1.value)").explain(true)
> {code}
> Which produces a similar plan, with a {{LeftSemi}} and no filters:
> {code:sql}
> == Physical Plan ==
> *(2) BroadcastHashJoin [value#19], [value#22], LeftSemi, BuildRight, false
> :- *(2) ColumnarToRow
> : +- FileScan parquet [value#19] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/tmp/my_test_table], PartitionFilters: [], PushedFilters: [], ReadSchema: struct<value:int>
> +- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0, int, true] as bigint)),false), [id=#125]
> +- *(1) ColumnarToRow
> +- FileScan parquet [value#22] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/tmp/my_test_table], PartitionFilters: [], PushedFilters: [], ReadSchema: struct<value:int>
> {code}
> If we naively add an extra optimizer batch for {{InferFiltersFromConstraints}}: https://github.com/abellina/spark/commit/8aaeb89151e04101c9513d7d7abd21cd00348acb, we get the desired physical plan:
> {code:sql}
> == Physical Plan ==
> *(2) BroadcastHashJoin [value#7], [value#13], LeftSemi, BuildRight, false
> :- *(2) Filter isnotnull(value#7)
> : +- *(2) ColumnarToRow
> : +- FileScan parquet [value#7] Batched: true, DataFilters: [isnotnull(value#7)], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/tmp/my_test_table], PartitionFilters: [], PushedFilters: [IsNotNull(value)], ReadSchema: struct<value:int>
> +- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0, int, false] as bigint)),false), [id=#146]
> +- *(1) Filter isnotnull(value#13)
> +- *(1) ColumnarToRow
> +- FileScan parquet [value#13] Batched: true, DataFilters: [isnotnull(value#13)], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/tmp/my_test_table], PartitionFilters: [], PushedFilters: [IsNotNull(value)], ReadSchema: struct<value:
> {code}
--
This message was sent by Atlassian Jira
(v8.20.7#820007)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org