Uploaded image for project: 'Spark'
  1. Spark
  2. SPARK-32859

Introduce SQL physical plan rule to decide enable/disable bucketing

    XMLWordPrintableJSON

    Details

    • Type: Improvement
    • Status: Resolved
    • Priority: Minor
    • Resolution: Fixed
    • Affects Version/s: 3.1.0
    • Fix Version/s: 3.1.0
    • Component/s: SQL
    • Labels:
      None

      Description

      Discussed with Wenchen Fan offline, it would be better that we can decide enable/disable SQL bucketing automatically according to query plan. Currently bucketing is enabled by default (`spark.sql.sources.bucketing.enabled`=true), so for all bucketed tables in the query plan, we will use bucket table scan (all input files per the bucket will be read by same task). This has the drawback that if the bucket table scan is not benefitting at all (no join/groupby/etc in the query), we don't need to use bucket table scan as it would restrict the # of tasks to be # of buckets and might hurt parallelism.

       

      The proposed change is to introduce a physical plan rule (right before `ensureRequirements`):

      (1).transformUp() physical plan, matching SparkPlan operator which is FileSourceScanExec, if optionalBucketSet is set, enabling bucket scan (bucket filter in this case).

      (2).transformUp() physical plan, matching SparkPlan operator which is SparkPlanWithInterestingPartitioning.

      SparkPlanWithInterestingPartitioning: the plan is in {SortMergeJoinExec, ShuffledHashJoinExec, HashAggregateExec, ObjectHashAggregateExec, SortAggregateExec, etc, which has HashClusteredDistribution/ClusteredDistribution in requiredChildDistribution}, and its requiredChildDistribution HashClusteredDistribution/ClusteredDistribution on its underlying FileSourceScanExec's bucketed columns.

      (3).for any child of SparkPlanWithInterestingPartitioning, which does not satisfy the plan's requiredChildDistribution: go though the child's sub query plan tree.
      if (3.1).all node's outputPartitioning is same as child, and all node's requiredChildDistribution is UnspecifiedDistribution.
      and (3.2).the leaf node is FileSourceScanExec on bucketed table and
      and (3.3).if enabling bucket scan for this FileSourceScanExec, the outputPartitioning of FileSourceScanExec satisfies requiredChildDistribution of SparkPlanWithInterestingPartitioning.
      If (3.1),(3.2),(3.3) are all true, enabling bucket scan for this FileSourceScanExec. And double check the new child of SparkPlanWithInterestingPartitioning satisfies requiredChildDistribution.

       

      The idea of SparkPlanWithInterestingPartitioning, is inspired from "interesting order" in "Access Path Selection in a Relational Database Management System"(http://www.inf.ed.ac.uk/teaching/courses/adbs/AccessPath.pdf).

        Attachments

          Issue Links

            Activity

              People

              • Assignee:
                chengsu Cheng Su
                Reporter:
                chengsu Cheng Su
              • Votes:
                0 Vote for this issue
                Watchers:
                10 Start watching this issue

                Dates

                • Created:
                  Updated:
                  Resolved: