[SPARK-31412] New Adaptive Query Execution in Spark SQL

• Type: New Feature
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 3.0.0
• Fix Version/s: 3.0.0
• Component/s: SQL
• Assignee: Unassigned
• Reporter: Wenchen Fan
• Created: 10/Apr/20 08:53
• Updated: 17/Dec/20 18:17
• Resolved: 10/Apr/20 09:32

Description

SPARK-9850 proposed the basic idea of adaptive execution in Spark. In DAGScheduler, a new API is added to support submitting a single map stage.  The current implementation of adaptive execution in Spark SQL supports changing the reducer number at runtime. An Exchange coordinator is used to determine the number of post-shuffle partitions for a stage that needs to fetch shuffle data from one or multiple stages. The current implementation adds ExchangeCoordinator while we are adding Exchanges. However there are some limitations. First, it may cause additional shuffles that may decrease the performance. We can see this from EnsureRequirements rule when it adds ExchangeCoordinator.  Secondly, it is not a good idea to add ExchangeCoordinators while we are adding Exchanges because we don’t have a global picture of all shuffle dependencies of a post-shuffle stage. I.e. for 3 tables’ join in a single stage, the same ExchangeCoordinator should be used in three Exchanges but currently two separated ExchangeCoordinator will be added. Thirdly, with the current framework it is not easy to implement other features in adaptive execution flexibly like changing the execution plan and handling skewed join at runtime.

We'd like to introduce a new way to do adaptive execution in Spark SQL and address the limitations. The idea is described at https://docs.google.com/document/d/1mpVjvQZRAkD-Ggy6-hcjXtBPiQoVbZGe3dLnAKgtJ4k/edit?usp=sharing

Issue Links

• causes: SPARK-33822 TPCDS Q5 fails if spark.sql.adaptive.enabled=true
• is related to: SPARK-33828 SQL Adaptive Query Execution QA
• relates to: SPARK-9850 Adaptive execution in Spark

Sub-Tasks

• The basic framework for the new Adaptive Query Execution: Resolved
• Adjust post shuffle partition number in adaptive execution: Resolved
• Disable OptimizeSkewJoin rule if introducing additional shuffle.: Resolved
• collect the runtime statistics of row count in map stage: Open
• Optimize skewed join at runtime with new Adaptive Execution: Resolved
• add metrics to AQE shuffle reader: Resolved
• add an individual config for skewed partition threshold: Resolved
• optimize skew join after shuffle partitions are coalesced: Resolved
• make skew join split skewed partitions more evenly: Resolved
• refine AQE config names: Resolved
• Dynamically reuse subqueries in AQE: Resolved
• Add a simple cost check for Adaptive Query Execution: Resolved
• Optimize shuffle reader to local shuffle reader when smj converted to bhj in adaptive execution: Resolved
• improve the splitting of skewed partitions: Resolved
• Add the user guide for Adaptive Query Execution: Resolved
• change the default value of minPartitionNum in AQE: Resolved
• Avoid changing SMJ to BHJ if the build side has a high ratio of empty partitions: Resolved
• Add tree traversal helper for adaptive spark plans: Resolved
• Optimize shuffle fetch of contiguous partition IDs: Resolved
• LocalShuffleReaderExec.outputPartitioning should use the corrected attributes: Resolved
• Improve the local reader performance by changing the task number from 1 to multi: Resolved
• Reading of csv file fails with adaptive execution turned on: Resolved
• Catch the exception when do materialize in AQE: Resolved
• Add adaptive execution context: Resolved
• remove ReusedQueryStageExec: Resolved
• Fix tests when enable Adaptive Query Execution: Resolved
• reset the metrics info of AdaptiveSparkPlanExec plan when enable aqe: Resolved
• Fix the NoSuchElementException exception when enable AQE with InSubquery use case: Resolved
• Fix the subquery metrics showing issue in UI When enable AQE: Resolved
• coalesce shuffle reader with splitting shuffle fetch request fails: Resolved
• AQE should not issue a "not supported" warning for queries being by-passed: Resolved
• Combine the skewed readers into one in AQE skew join optimizations: Resolved
• Subqueries should not be AQE-ed if main query is not: Resolved
• Turning off AQE in CacheManager is not thread-safe: Resolved
• Refactor AQE readers and RDDs: Resolved
• Remove the max split config after changing the multi sub joins to multi sub partitions: Resolved
• Don't cancel a QueryStageExec when it's already finished: Resolved
• Add config for AQE logging level: Resolved
• Make more efficient and clean up AQE update UI code: Resolved
• Replace `Array` with `Seq` in AQE `CustomShuffleReaderExec`: Resolved
• AQE will use the same SubqueryExec even if subqueryReuseEnabled=false: Resolved
• NPE in OptimizeSkewedJoin when there's a inputRDD of plan has 0 partition: Resolved
• SQL UI doesn't show write commands of AQE plan: Resolved
• The final AdaptiveSparkPlan event is not marked with `isFinalPlan=true`: Resolved