You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by gu...@apache.org on 2021/06/23 00:55:18 UTC
[spark] branch branch-3.1 updated: [SPARK-35695][SQL][FOLLOWUP] Use
AQE helper to simplify the code in CollectMetricsExec
This is an automated email from the ASF dual-hosted git repository.
gurwls223 pushed a commit to branch branch-3.1
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.1 by this push:
new d1b51d7 [SPARK-35695][SQL][FOLLOWUP] Use AQE helper to simplify the code in CollectMetricsExec
d1b51d7 is described below
commit d1b51d7a0b126ec4657986b0b5bea069dfbf0887
Author: Wenchen Fan <we...@databricks.com>
AuthorDate: Wed Jun 23 09:54:12 2021 +0900
[SPARK-35695][SQL][FOLLOWUP] Use AQE helper to simplify the code in CollectMetricsExec
### What changes were proposed in this pull request?
This is a followup of https://github.com/apache/spark/pull/32862 , to simplify the code with AQE helper.
### Why are the changes needed?
code cleanup
### Does this PR introduce _any_ user-facing change?
no
### How was this patch tested?
existing tests
Closes #33026 from cloud-fan/follow.
Authored-by: Wenchen Fan <we...@databricks.com>
Signed-off-by: Hyukjin Kwon <gu...@apache.org>
(cherry picked from commit a87ee5d8b9dc00e327edc9911c21225c09042acd)
Signed-off-by: Hyukjin Kwon <gu...@apache.org>
---
.../org/apache/spark/sql/execution/CollectMetricsExec.scala | 13 +++++--------
1 file changed, 5 insertions(+), 8 deletions(-)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/CollectMetricsExec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/CollectMetricsExec.scala
index 7a59504..c9e1bc3 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/CollectMetricsExec.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/CollectMetricsExec.scala
@@ -22,7 +22,7 @@ import org.apache.spark.sql.Row
import org.apache.spark.sql.catalyst.{CatalystTypeConverters, InternalRow}
import org.apache.spark.sql.catalyst.expressions.{Attribute, NamedExpression, SortOrder}
import org.apache.spark.sql.catalyst.plans.physical.Partitioning
-import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanExec, QueryStageExec}
+import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
import org.apache.spark.sql.execution.columnar.InMemoryTableScanExec
import org.apache.spark.sql.types.StructType
@@ -86,19 +86,16 @@ case class CollectMetricsExec(
}
}
-object CollectMetricsExec {
+object CollectMetricsExec extends AdaptiveSparkPlanHelper {
/**
* Recursively collect all collected metrics from a query tree.
*/
def collect(plan: SparkPlan): Map[String, Row] = {
- val metrics = plan.collectWithSubqueries {
- case collector: CollectMetricsExec => Map(collector.name -> collector.collectedMetrics)
+ val metrics = collectWithSubqueries(plan) {
+ case collector: CollectMetricsExec =>
+ Map(collector.name -> collector.collectedMetrics)
case tableScan: InMemoryTableScanExec =>
CollectMetricsExec.collect(tableScan.relation.cachedPlan)
- case adaptivePlan: AdaptiveSparkPlanExec =>
- CollectMetricsExec.collect(adaptivePlan.executedPlan)
- case queryStageExec: QueryStageExec =>
- CollectMetricsExec.collect(queryStageExec.plan)
}
metrics.reduceOption(_ ++ _).getOrElse(Map.empty)
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org