remove metrics, because they are very slow

apache · Jan 19, 2016 · 331b46d · 331b46d
1 parent f0f3da6
commit 331b46d
Show file tree

Hide file tree

Showing 3 changed files with 5 additions and 30 deletions.
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/WholeStageCodegen.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/WholeStageCodegen.scala
@@ -22,10 +22,9 @@ import scala.collection.mutable.ArrayBuffer
 import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.SQLContext
 import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.expressions.{Attribute, BoundReference, Expression, LeafExpression}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, BoundReference, LeafExpression}
 import org.apache.spark.sql.catalyst.expressions.codegen._
 import org.apache.spark.sql.catalyst.rules.Rule
-import org.apache.spark.sql.execution.metric.LongSQLMetric
 
 /**
   * An interface for those physical operators that support codegen.
@@ -91,21 +90,9 @@ trait CodegenSupport extends SparkPlan {
     *   }
     */
   def doConsume(ctx: CodegenContext, child: SparkPlan, input: Seq[ExprCode]): String
-
-  /**
-    * Return a term for a LongSQLMetric specified by given name.
-    */
-  protected def termForLongMetric(ctx: CodegenContext, name: String): String = {
-    val metric = longMetric(name)
-    val idx = ctx.references.length
-    ctx.references += metric
-    val term = ctx.freshName(name)
-    val clsName = classOf[LongSQLMetric].getName
-    ctx.addMutableState(clsName, term, s"$term = ($clsName) references[$idx];")
-    term
-  }
 }
 
+
 /**
   * InputAdapter is used to hide a SparkPlan from a subtree that support codegen.
   *

diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/basicOperators.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/basicOperators.scala
@@ -24,7 +24,7 @@ import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext, ExprCode, ExpressionCanonicalizer}
 import org.apache.spark.sql.catalyst.plans.physical._
-import org.apache.spark.sql.execution.metric.{LongSQLMetric, SQLMetrics}
+import org.apache.spark.sql.execution.metric.SQLMetrics
 import org.apache.spark.sql.types.LongType
 import org.apache.spark.util.MutablePair
 import org.apache.spark.util.random.PoissonSampler
@@ -42,14 +42,11 @@ case class Project(projectList: Seq[NamedExpression], child: SparkPlan)
   }
 
   override def doConsume(ctx: CodegenContext, child: SparkPlan, input: Seq[ExprCode]): String = {
-    val numRows = termForLongMetric(ctx, "numRows")
     val exprs = projectList.map(x =>
       ExpressionCanonicalizer.execute(BindReferences.bindReference(x, child.output)))
     ctx.currentVars = input
     val output = exprs.map(_.gen(ctx))
     s"""
-       | // Projection
-       | $numRows.add(1);
        | ${output.map(_.code).mkString("\n")}
        |
        | ${consume(ctx, output)}
@@ -84,18 +81,13 @@ case class Filter(condition: Expression, child: SparkPlan) extends UnaryNode wit
   }
 
   override def doConsume(ctx: CodegenContext, child: SparkPlan, input: Seq[ExprCode]): String = {
-    val numInputTerm = termForLongMetric(ctx, "numInputRows")
-    val numOutputTerm = termForLongMetric(ctx, "numOutputRows")
     val expr = ExpressionCanonicalizer.execute(
       BindReferences.bindReference(condition, child.output))
     ctx.currentVars = input
     val eval = expr.gen(ctx)
     s"""
-       | // Filter
-       | $numInputTerm.add(1);
        | ${eval.code}
        | if (!${eval.isNull} && ${eval.value}) {
-       |   $numOutputTerm.add(1);
        |   ${consume(ctx, ctx.currentVars)}
        | }
      """.stripMargin

diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/metric/SQLMetricsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/metric/SQLMetricsSuite.scala
@@ -135,13 +135,9 @@ class SQLMetricsSuite extends SparkFunSuite with SharedSQLContext {
   test("WholeStageCodegen metrics") {
     // Assume the execution plan is
     // WholeStageCodegen(nodeId = 0, Range(nodeId = 2) -> Filter(nodeId = 1))
+    // TODO: update metrics in generated operators
     val df = sqlContext.range(10).filter('id < 5)
-    testSparkPlanMetrics(df, 1, Map(
-      1L -> ("Filter", Map(
-        "number of input rows" -> 10L,
-        "number of output rows" -> 5L
-      )))
-    )
+    testSparkPlanMetrics(df, 1, Map.empty)
   }
 
   test("TungstenAggregate metrics") {