Minor Description
Discussed with cloud_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
- Dependency
-
SPARK-24528 Add support to read multiple sorted bucket files for data source v1
-
- In Progress
-
- links to
