[
https://issues.apache.org/jira/browse/SPARK-30528?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17020385#comment-17020385
]
Wei Xue commented on SPARK-30528:
---------------------------------
Good point, [~mayurb31]!
1. Heuristics: yes, we should improve the cost estimate for the filter plan. As
a quick workaround, though, you could set
"spark.sql.optimizer.dynamicPartitionPruning.fallbackFilterRatio" to "0.0",
which would disable this kind of DPP if columns stats are not available.
2. Reuse: yes, that's a dilemma here: first of all we wanna reduce the result
set returned to the driver for pruning values, and that's why the Aggregate is
added. It might kill some potential opportunities for exchange reuse if the
filter plan contains another join, but that kind of potential opportunity is
not fully guaranteed even if we didn't push down the Aggregate, for the join
in the filter plan can turn out to be a BHJ. However, `pruningHasBenefit` had
intended to cost the entire DPP subquery as overhead without considering reuse,
so this takes us back to point 1: we should improve the costing of heuristics
so that this kind of DPP should not be triggered at all if the scan + join
would be just too much work.
3. Can you attach the query plan instead of UI for your last example? I think
the reuse did not happen because the first subquery selects `col1` while the
second `col2`?
> DPP issues
> ----------
>
> Key: SPARK-30528
> URL: https://issues.apache.org/jira/browse/SPARK-30528
> Project: Spark
> Issue Type: Bug
> Components: Optimizer
> Affects Versions: 3.0.0
> Reporter: Mayur Bhosale
> Priority: Major
> Labels: performance
> Attachments: cases.png, dup_subquery.png, plan.png
>
>
> In DPP, heuristics to decide if DPP is going to benefit relies on the sizes
> of the tables in the right subtree of the join. This might not be a correct
> estimate especially when the detailed column level stats are not available.
> {code:java}
> // the pruning overhead is the total size in bytes of all scan relations
> val overhead =
> otherPlan.collectLeaves().map(_.stats.sizeInBytes).sum.toFloat
> filterRatio * partPlan.stats.sizeInBytes.toFloat > overhead.toFloat
> {code}
> Also, DPP executes the entire right side of the join as a subquery because of
> which multiple scans happen for the tables in the right subtree of the join.
> This can cause issues when join is non-Broadcast Hash Join (BHJ) and reuse of
> the subquery result does not happen. Also, I couldn’t figure out, why do the
> results from the subquery get re-used only for BHJ?
>
> Consider a query,
> {code:java}
> SELECT *
> FROM store_sales_partitioned
> JOIN (SELECT *
> FROM store_returns_partitioned,
> date_dim
> WHERE sr_returned_date_sk = d_date_sk) ret_date
> ON ss_sold_date_sk = d_date_sk
> WHERE d_fy_quarter_seq > 0
> {code}
> DPP will kick-in for both the join. (Please check the image plan.png attached
> below for the plan)
> Some of the observations -
> * Based on heuristics, DPP would go ahead with pruning if the cost of
> scanning the tables in the right sub-tree of the join is less than the
> benefit due to pruning. This is due to the reason that multiple scans will be
> needed for an SMJ. But heuristics simply checks if the benefits offset the
> cost of multiple scans and do not take into consideration other operations
> like Join, etc in the right subtree which can be quite expensive. This issue
> will be particularly prominent when detailed column level stats are not
> available. In the example above, a decision that pruningHasBenefit was made
> on the basis of sizes of the tables store_returns_partitioned and date_dim
> but did not take into consideration the join between them before the join
> happens with the store_sales_partitioned table.
> * Multiple scans are needed when the join is SMJ as the reuse of the
> exchanges does not happen. This is because Aggregate gets added on top of the
> right subtree to be executed as a subquery in order to prune only required
> columns. Here, scanning all the columns as the right subtree of the join
> would, and reusing the same exchange might be more helpful as it avoids
> duplicate scans.
> This was just a representative example, but in-general for cases such as in
> the image cases.png below, DPP can cause performance issues.
>
> Also, for the cases when there are multiple DPP compatible join conditions in
> the same join, the entire right subtree of the join would be executed as a
> subquery that many times. Consider an example,
> {code:java}
> SELECT *
> FROM partitionedtable
> JOIN nonpartitionedtable
> ON partcol1 = col1
> AND partcol2 = col2
> WHERE nonpartitionedtable.id > 0
> {code}
> Here the right subtree of the join (scan of table nonpartitionedtable) would
> be executed twice as a subquery, once each for the every join condition.
> These two subqueries should be aggregated and executed only once as they are
> almost the same apart from the columns that they prune. Check the image
> dup_subquery.png attached below for the details.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
