Uh oh!
There was an error while loading. Please reload this page.
[SPARK-15616][SQL] Hive table supports partition pruning in JoinSelection - #25919
[SPARK-15616][SQL] Hive table supports partition pruning in JoinSelection#25919advancedxy wants to merge 8 commits into
Conversation
advancedxy
commented
Sep 24, 2019
cc @cloud-fan. |
cloud-fan
commented
Sep 24, 2019
ok to test |
cloud-fan
commented
Sep 24, 2019
add to whitelist |
| predicate.references.subsetOf(partitionSet) | ||
| } | ||
| val conf = session.sessionState.conf | ||
| if (pruningPredicates.nonEmpty && conf.fallBackToHdfsForStatsEnabled && |
There was a problem hiding this comment.
Why we need to check conf.fallBackToHdfsForStatsEnabled?
There was a problem hiding this comment.
We should only get size from HDFS if conf.fallBackToHdfsForStatsEnabled? Since it could be a time-consuming operation.
Though, this condition should probably be pushed down to before the CommandUtils.calculateLocationSize call
Uh oh!
There was an error while loading. Please reload this page.
SparkQA
commented
Sep 24, 2019
Test build #111297 has finished for PR 25919 at commit
|
advancedxy
commented
Oct 20, 2019
@cloud-fan Now pruned partitions are cached in HiveTableRelation, what do you think about current approach ? |
SparkQA
commented
Oct 20, 2019
Test build #112331 has finished for PR 25919 at commit
|
SparkQA
commented
Oct 20, 2019
Test build #112332 has finished for PR 25919 at commit
|
SparkQA
commented
Oct 20, 2019
Test build #112334 has finished for PR 25919 at commit
|
SparkQA
commented
Oct 20, 2019
Test build #112337 has finished for PR 25919 at commit
|
SparkQA
commented
Oct 21, 2019
Test build #112356 has finished for PR 25919 at commit
|
| tableStats: Option[Statistics] = None) extends LeafNode with MultiInstanceRelation { | ||
| tableStats: Option[Statistics] = None, | ||
| @transient normalizedFilters: Seq[Expression] = Nil, | ||
| @transient prunedPartitions: Seq[CatalogTablePartition] = Nil) |
There was a problem hiding this comment.
How can we distinguish 0 partitions after pruning, and not being partition pruned?
There was a problem hiding this comment.
We have another field called normalizedFilters, when it's empty(Nil), then the prunedPartitions are not pruned, otherwise it could be 0 partitions after pruning when prunedPartitions = Nil
| case class PruneHiveTablePartitions( | ||
| session: SparkSession) extends Rule[LogicalPlan] with PredicateHelper { | ||
| override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators { | ||
| case filter @ Filter(condition, relation: HiveTableRelation) if relation.isPartitioned => |
There was a problem hiding this comment.
can we follow PruneFileSourcePartitions? I think we should also support Filter(Project(HiveScan))
| normalizedFilters) | ||
| val isFiltersEqual = normalizedFilters.zip(relation.normalizedFilters) | ||
| .forall { case (e1, e2) => e1.semanticEquals(e2) } | ||
| if (isFiltersEqual) { |
There was a problem hiding this comment.
Only under exactly matched pruning filters, we can simply get partitions from HiveTableRelation
| val withStats = relation.tableMeta.copy( | ||
| stats = Some(CatalogStatistics(sizeInBytes = BigInt(sizeInBytes)))) | ||
| val prunedHiveTableRelation = relation.copy(tableMeta = withStats, | ||
| normalizedFilters = pruningPredicates, prunedPartitions = prunedPartitions) |
There was a problem hiding this comment.
Why do we need to keep pruningPredicates? IIUC the approach should be very simply:
- this rule only changes
HiveTableRelationto hold an optional partition list. - the
HiveTableScanExecwill get the partition list fromHiveTableRelationor calllistPartitionsByFilter.
There was a problem hiding this comment.
Due to SPARK-24085, the pruningPredicates(we eliminate the subquery) could be different than the filters passed to HiveTableScan. So I keep the pruningPredicates, and only retrieves the prunedPartitions when HiveTableScanExec's pruningPartitionPredict matches exactly with HiveTableRelation's normalizedFilters.
The simplified solution occurred to me first, then I thought the filters could be different for some reason, and SPARK-24085 is an example, hence the proposed solution here.
1. don't store pruningFilters in HiveTableRelation 2. follow PruneFilSourcePartitions's style to extract projections, predicates and hive relation 3. skip partition pruning if scalar subquery is involved.
SparkQA
commented
Oct 25, 2019
Test build #112679 has finished for PR 25919 at commit
|
SparkQA
commented
Oct 26, 2019
Test build #112707 has finished for PR 25919 at commit
|
advancedxy
commented
Oct 26, 2019
retest it please |
advancedxy
commented
Nov 22, 2019
Gently ping @cloud-fan |
maropu
commented
Nov 24, 2019
retest this please |
maropu
commented
Nov 24, 2019
still |
SparkQA
commented
Nov 24, 2019
Test build #114331 has finished for PR 25919 at commit
|
advancedxy
commented
Nov 25, 2019
I think it's ready for review. |
| val normalizedFilters = partitionPruningPred.map(_.transform { | ||
| case a: AttributeReference => originalAttributes(a) | ||
| }) | ||
| sparkSession.sessionState.catalog.listPartitionsByFilter( |
There was a problem hiding this comment.
@cloud-fan@maropu@advancedxy
Since the rawPartitions are called by "prunePartitions(rawPartitions)" in doExecute method, it seems prunePartitions will filter out all irrelevant partitions using "boundPruningPred". Then why we still need to call listpartitionsByFilter here ?
Could you please help me understand this ? thanks a lot in advance.
| !predicate.references.isEmpty && predicate.references.subsetOf(partitionSet) | ||
| } | ||
| // SPARK-24085: scalar subquery should be skipped for partition pruning | ||
| val hasScalarSubquery = pruningPredicates.exists(SubqueryExpression.hasSubquery) |
There was a problem hiding this comment.
It skips all subqueries instead of scalar subqueries.
| rawDataSize.get | ||
| } else if (totalSize.isDefined && totalSize.get > 0L) { | ||
| totalSize.get | ||
| } else if (conf.fallBackToHdfsForStatsEnabled) { |
There was a problem hiding this comment.
Per the doc of the conf "spark.sql.statistics.fallBackToHdfs", it is only for non-partitioned hive table :
"This flag is effective only for non-partitioned Hive tables."
advancedxy
commented
Jan 8, 2020
closed in favor of #26805 |
### What changes were proposed in this pull request? Add optimizer rule PruneHiveTablePartitions pruning hive table partitions based on filters on partition columns. Doing so, the total size of pruned partitions may be small enough for broadcast join in JoinSelection strategy. ### Why are the changes needed? In JoinSelection strategy, spark use the "plan.stats.sizeInBytes" to decide whether the plan is suitable for broadcast join. Currently, "plan.stats.sizeInBytes" does not take "pruned partitions" into account, so it may miss some broadcast join and take sort-merge join instead, which will definitely impact join performance. This PR aim at taking "pruned partitions" into account for hive table in "plan.stats.sizeInBytes" and then improve performance by using broadcast join if possible. ### Does this PR introduce any user-facing change? no ### How was this patch tested? Added unit tests. This is based on #25919, credits should go to lianhuiwang and advancedxy. Closes#26805 from fuwhu/SPARK-15616. Authored-by: fuwhu <bestwwg@163.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com>
Add optimizer rule PruneHiveTablePartitions pruning hive table partitions based on filters on partition columns. Doing so, the total size of pruned partitions may be small enough for broadcast join in JoinSelection strategy. In JoinSelection strategy, spark use the "plan.stats.sizeInBytes" to decide whether the plan is suitable for broadcast join. Currently, "plan.stats.sizeInBytes" does not take "pruned partitions" into account, so it may miss some broadcast join and take sort-merge join instead, which will definitely impact join performance. This PR aim at taking "pruned partitions" into account for hive table in "plan.stats.sizeInBytes" and then improve performance by using broadcast join if possible. no Added unit tests. This is based on apache#25919, credits should go to lianhuiwang and advancedxy. Closesapache#26805 from fuwhu/SPARK-15616. Authored-by: fuwhu <bestwwg@163.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com>
What changes were proposed in this pull request?
A new optimizer strategy called
PruneHiveTablePartitionsis added, which calculates table size as the total size of pruned partitions. Thus, Spark planner can pick upBroadcastJoinif the size of pruned partitions is under broadcast join threshold.Why are the changes needed?
This is a performance improvement.
Does this PR introduce any user-facing change?
No.
How was this patch tested?
Added unit tests.
This is based on #18193, credits should go to @lianhuiwang.